airflow_dag_pagerduty_create_incident_operator_example.py

1from datetime import datetime, timedelta
2from airflow import DAG
3from airflow.providers.pagerduty.operators.pagerduty import PagerDutyCreateIncidentOperator
4
5with DAG(
6    "pagerduty_incident_example",
7    start_date=datetime(2023, 1, 1),
8    schedule_interval=None,
9    catchup=False,
10    default_args={
11        "pagerduty_conn_id": "pagerduty_default",
12    },
13) as dag:
14
15    # Create a PagerDuty incident
16    create_incident = PagerDutyCreateIncidentOperator(
17        task_id="create_pagerduty_incident",
18        summary="Airflow DAG failure alert",
19        severity="critical",
20        source="Airflow Worker 01",
21        component="ETL Pipeline",
22        group="Data Engineering",
23        class_type="incident",
24        custom_details={
25            "dag_id": "{{ dag.dag_id }}",
26            "run_id": "{{ run_id }}",
27            "execution_date": "{{ ds }}",
28        },
29    )
30
31    create_incident