PagerDuty Airflow Provider

Warnings

• breaking Provider version 5.0.0 removed all deprecated classes, parameters, and features. Ensure your code does not rely on any previously deprecated components.
  Fix: Refer to the official changelog for specific removals. Update your code to use current API methods.
• breaking Minimum Apache Airflow version required is 2.11.0. Older Airflow versions are not supported.
  Fix: Upgrade your Apache Airflow environment to at least version 2.11.0. If upgrading from Airflow <2.1.0, an `airflow upgrade db` might be necessary.
• breaking Python 3.9 support was dropped with version 5.0.1. Ensure your environment uses Python 3.10 or newer.
  Fix: Upgrade your Python environment to version 3.10 or later.
• gotcha When configuring a PagerDuty connection in Airflow, there are two main types: 'Pagerduty' for the REST API (requires an API token in the 'Password' field) and 'Pagerduty Events' for the Events API (requires an Integration Key). Using the wrong type or credential will lead to authentication failures.
  Fix: Select the correct connection type in the Airflow UI (Admin -> Connections) and provide the corresponding API token or Integration Key in the designated field. For Events API, the integration key can often be passed directly to the `send_pagerduty_notification` function.
• gotcha An issue (reported in Jan 2025 as #45626) indicated that the `PagerdutyNotifier` class's `notify` method was attempting to use a `create_event` method that no longer existed, leading to failed notifications. While this might be resolved in current versions, using the `send_pagerduty_notification` function directly in callbacks is often a more robust and commonly demonstrated approach.
  Fix: Prefer using the `send_pagerduty_notification` function for DAG/task callbacks. If using `PagerdutyNotifier`, ensure your provider version is sufficiently updated and test the notification flow thoroughly.

Install

• pip install apache-airflow-providers-pagerduty Install latest version

Imports

• PagerdutyHook

  from airflow.providers.pagerduty.hooks.pagerduty import PagerdutyHook

• send_pagerduty_notification

  from airflow.providers.pagerduty.notifications.pagerduty import send_pagerduty_notification

• PagerdutyNotifier

  from airflow.providers.pagerduty.notifications.pagerduty import PagerdutyNotifier

  Import from the specific 'pagerduty' module within 'notifications'.

Quickstart

This quickstart demonstrates how to use `send_pagerduty_notification` as an `on_failure_callback` for both a DAG and a specific task. Ensure you have configured a PagerDuty connection in Airflow (Admin -> Connections, type 'Pagerduty' or 'Pagerduty Events') or provide the `integration_key` directly. For security, it's recommended to store sensitive keys in Airflow Connections or a secret backend, and retrieve them via `BaseHook.get_connection`.

import os
from datetime import datetime

from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.providers.pagerduty.notifications.pagerduty import send_pagerduty_notification

PAGERDUTY_INTEGRATION_KEY = os.environ.get('PAGERDUTY_INTEGRATION_KEY', '')

with DAG(
    dag_id='pagerduty_notification_example',
    start_date=datetime(2023, 1, 1),
    schedule=None,
    catchup=False,
    tags=['pagerduty', 'notification'],
    on_failure_callback=[
        send_pagerduty_notification(
            summary="The DAG {{ dag.dag_id }} failed!",
            severity="critical",
            source="airflow dag_id: {{dag.dag_id}}",
            dedup_key="{{dag.dag_id}}-failure",
            component="airflow",
            integration_key=PAGERDUTY_INTEGRATION_KEY,
        )
    ]
) as dag:
    start_task = BashOperator(
        task_id='start_task',
        bash_command='echo "Starting DAG..."',
    )

    failing_task = BashOperator(
        task_id='failing_task',
        bash_command='exit 1', # This task will intentionally fail
        on_failure_callback=[
            send_pagerduty_notification(
                summary="Task {{ ti.task_id }} failed in DAG {{ dag.dag_id }}",
                severity="error",
                source="airflow task_id: {{ti.task_id}}",
                dedup_key="{{dag.dag_id}}-{{ti.task_id}}-failure",
                component="airflow",
                integration_key=PAGERDUTY_INTEGRATION_KEY,
            )
        ]
    )

    success_task = BashOperator(
        task_id='success_task',
        bash_command='echo "DAG succeeded!"',
    )

    start_task >> failing_task >> success_task