Part 7

Alerting & Notifications

All four channels below are configured the same way: a plain dict under params={"_callbacks": {event: {...}}} on a task (feature 21) — event is one of on_success, on_failure, on_retry, on_skipped.

41Email alerts#

Parameters (config dict):

Field	Type	Default	Notes
`type`	string	required	`"email"`
`to`	list of strings	required	Recipient addresses.
`subject`	string	required	Supports 4 literal tokens (see below).
`html_content`	string	`""`	Supports the same 4 tokens.

How it works:

to and subject are both required — there's no default subject line.
subject and html_content support exactly four literal placeholders: {{dag_id}}, {{task_id}}, {{run_id}}, {{event}} (with or without inner spaces) — this is a simple find-and-replace, not a full templating engine, so Jinja/Go-style expressions won't render here.
Requires SMTP to be configured for the deployment — if it isn't, the alert is silently not sent (the callback is still marked processed).
Fires per task instance, once per triggering event — a mapped task (feature 32) sends its own email per map_index that reaches the configured event.

Example:

python

notify_failure = PythonOperator(
    task_id="risky_step",
    python_callable=lambda: None,
    params={
        "_callbacks": {
            "on_failure": {
                "type": "email",
                "to": ["oncall@company.com"],
                "subject": "PI-Flow failure: {{dag_id}}.{{task_id}}",
                "html_content": "<p>Run <b>{{run_id}}</b> failed on event {{event}}.</p>",
            }
        }
    },
)

42Slack alerts#

Parameters (config dict):

Field	Type	Default	Notes
`type`	string	required	`"slack"`
`connection_id`	string	required	Must be a webhook-mode Slack connection.

How it works:

Use a Slack connection created in webhook mode specifically for alerting. If your primary Slack connection uses a bot token (for SlackAPIPostOperator task usage, feature 60), create a separate webhook-mode connection for this callback channel.
The message content isn't configurable here — it's a fixed-format summary (DAG/Task/Run/Event). If you need custom message text, use the HTTP webhook channel (feature 43) or a dedicated SlackAPIPostOperator task instead (feature 60).
A non-200 response from Slack is treated as a failed send and logged — there's no automatic retry for a transient Slack outage.

Example:

python

notify_failure = PythonOperator(
    task_id="risky_step",
    python_callable=lambda: None,
    params={
        "_callbacks": {
            "on_failure": {
                "type": "slack",
                "connection_id": "slack_alerts_webhook",  # must be webhook-mode
            }
        }
    },
)

43HTTP webhook alerts#

Parameters (config dict):

Field	Type	Default	Notes
`type`	string	required	`"http_webhook"`
`url`	string	required	Supports the same 4 tokens as email.
`headers`	dict	`{}`	Put any auth headers here — no connection lookup for this channel.
`body`	string	a default JSON payload	Supports the same 4 tokens.

How it works:

Both url and body support the same 4-token replacement as email ({{dag_id}}, {{task_id}}, {{run_id}}, {{event}}) — you can parameterize the URL path itself, not just the body.
There's no connection lookup for this channel — bake any bearer token/API key directly into headers.
Any 2xx response counts as success; anything else is treated as a failed delivery.
If you omit body, a default JSON payload (dag_id, task_id, run_id, map_index, event, message, timestamp) is sent instead.

Example:

python

notify_failure = PythonOperator(
    task_id="risky_step",
    python_callable=lambda: None,
    params={
        "_callbacks": {
            "on_failure": {
                "type": "http_webhook",
                "url": "https://ops.company.com/hooks/{{dag_id}}",
                "headers": {"Authorization": "Bearer secret-token-value"},
                "body": '{"dag": "{{dag_id}}", "task": "{{task_id}}", "event": "{{event}}"}',
            }
        }
    },
)

44PagerDuty alerts#

Parameters (config dict):

Field	Type	Default	Notes
`type`	string	required	`"pagerduty"`
`connection_id`	string	required	Routing key stored under `extra.routing_key`, or as the connection's `password` field.

How it works:

Severity is automatically derived from the event: on_failure → critical, on_retry → warning, on_success → info. This mapping isn't configurable per task.
If you need custom severity logic, use the HTTP webhook channel (feature 43) directly against PagerDuty's Events API instead.
Every alert opens a new incident (event_action: "trigger") — there's no built-in "resolve" call to auto-close an incident when a retried task later succeeds; wire that up yourself as a downstream task if you need it.

Example:

python

notify_failure = PythonOperator(
    task_id="risky_step",
    python_callable=lambda: None,
    params={
        "_callbacks": {
            "on_failure": {
                "type": "pagerduty",
                "connection_id": "pagerduty_prod",
            }
        }
    },
)

45Event scope#

What it does: Explains exactly when each callback event fires, at both the task level (feature 21) and the DAG level (feature 9).

How it works:

on_failure fires once, on the task's terminal (final) failure — not on every retry attempt.
on_retry fires once per attempt that's about to be retried.
on_skipped only fires for the task's own terminal skip (PiFlowSkip, or a soft-fail sensor timeout) — it does not fire for a task skipped by branch evaluation (feature 33) or trigger-rule skip-propagation (feature 14), since those tasks never actually executed.
At the DAG level, only on_success, on_failure, and on_sla_miss exist — there's no DAG-level retry or skip event. If you need alerting on a partial-failure shape across the whole DAG, attach task-level callbacks to the specific tasks you care about, or add a dedicated notifier task.
Every event fires per task instance, including per map_index for a mapped task — there's no built-in "fire once for the whole mapped group" aggregation.
For a DAG-wide "final outcome, whatever it was" alert, add a dedicated notifier task rather than relying on a callback:

Example:

python

with DAG(dag_id="alerting_scope_demo", schedule="@daily", start_date=datetime(2026, 1, 1)) as dag:

    flaky = PythonOperator(
        task_id="flaky_step",
        python_callable=lambda: None,
        retries=3,
        params={
            "_callbacks": {
                # Fires once, on the FINAL failed attempt only
                "on_failure": {"type": "slack", "connection_id": "slack_alerts_webhook"},
                # Fires only if THIS task raises PiFlowSkip itself
                "on_skipped": {"type": "email", "to": ["team@company.com"], "subject": "skipped"},
            }
        },
    )

    # DAG-wide "final outcome" alert — a dedicated task, not a callback
    notify_final = SlackAPIPostOperator(
        task_id="notify_final",
        slack_conn_id="slack_alerts_webhook",
        text="alerting_scope_demo finished: check grid view for final status",
        trigger_rule="all_done",
    )

    flaky >> notify_final