Documentation Grafana Cloud Platform Workflows Reference Blueprints Learn: shape data with Python

Learn: shape data with Python

A runnable example of the Python step. Reshapes a list of alerts, finds noisy services, and passes the result to the next step as JSON. No setup needed.

This Grafana Workflows blueprint is in the Basics category of the Blueprints library.

Before you begin

This blueprint needs no integrations or secrets.

Steps

Steps run in order from first to last. A step inside a switch branch runs only when Workflows chooses that branch.

For the inputs and outputs of each step type, refer to Step and trigger types reference.

Before you turn on the workflow, customize any steps that need input from you, such as a channel ID or a schedule name. Each step’s description says what to set.

Workflow definition

Save this definition as a YAML file to create the workflow with gcx, the Workflows API, or Terraform, or give it to a coding agent as an example. Replace FOLDER_UID with the UID of the folder for the workflow. Blueprints start disabled, so finish customizing the steps before you set enabled to true. For details, refer to Manage workflows as code.

YAML Copy

apiVersion: workflows.ext.grafana.app/v1alpha1
kind: Definition
metadata:
  name: learn-python-data-shaping
  annotations:
    grafana.app/folder: <FOLDER_UID>
spec:
  name: 'Learn: shape data with Python'
  enabled: false
  startWhen:
    matchingRules: []
    schedules: []
  steps:
    - id: shape
      type: code.execute
      name: Group alerts by service
      description: Counts alerts per service and flags services at or above noisyThreshold. Change the threshold in args.
      inputs:
        language: 'python'
        args: '{"noisyThreshold": 2}'
        code: |
          import json
          from collections import Counter

          alerts = context["inputs"].get("data", {}).get("alerts", [])
          per_service = Counter(alert.get("service", "unknown") for alert in alerts)
          threshold = args.get("noisyThreshold", 2)

          print(json.dumps({
              "total": len(alerts),
              "perService": dict(per_service),
              "noisyServices": sorted(s for s, n in per_service.items() if n >= threshold),
          }))
    - id: check-success
      type: filter
      name: Fail if the script failed
      description: Fails the run when the Python code raised an exception. Check the stderr and error outputs for details.
      inputs:
        condition: 'steps.shape.outputs.success == true'
        failOnExit: 'true'
    - id: report
      type: transform
      name: Read the Python output
      description: Parses the script's stdout with json_parse() and reads fields from it.
      inputs:
        template: "${string(json_parse(steps.shape.outputs.stdout).total)} alerts. Noisy services: ${json_stringify(json_parse(steps.shape.outputs.stdout).noisyServices)}"

How it works

This section explains when the workflow runs and how its steps work together.

When it runs

This blueprint has no event or schedule trigger, so it runs only when you start it manually.

How the steps fit together

These notes cover how the steps connect and which Workflows behaviors the blueprint relies on.

• Manual-only workflow that needs no integrations.
• Suggested test payload data: {“alerts”: [{“service”: “checkout”, “severity”: “critical”}, {“service”: “checkout”, “severity”: “warning”}, {“service”: “search”, “severity”: “warning”}, {“service”: “checkout”, “severity”: “critical”}]}.
• It shows the code.execute contract.
• The workflow context is available as a dict named context, with inputs, steps, and resources (secrets are removed).
• args is a JSON string input, deserialized into a dict named args, and its values can contain ${...} expressions.
• The script prints a single JSON object to stdout, and later steps read it with json_parse(steps.<id>.outputs.stdout).
• A filter with failOnExit turns a Python exception (success false) into a failed run.
• The sandbox has no network access, so pair it with http.call to process API responses.

Next steps

• Blueprints
• Start from a blueprint
• Manage workflows as code
• Step and trigger types reference