airflow_dag_pagerduty_notifier_on_failure_callback.py

1from datetime import datetime
2from airflow import DAG
3from airflow.providers.pagerduty.notifications.pagerduty import send_pagerduty_notification
4from airflow.operators.empty import EmptyOperator
5
6with DAG(
7    dag_id="pagerduty_notifier_example",
8    start_date=datetime(2023, 1, 1),
9    schedule=None,
10    on_failure_callback=[
11        send_pagerduty_notification(
12            summary="The DAG {{ dag.dag_id }} failed",
13            severity="critical",
14            source="airflow-notification",
15            pagerduty_conn_id="pagerduty_default",
16        )
17    ],
18    catchup=False,
19):
20    EmptyOperator(task_id="task1")