Opsgenie 警报通知器¶

使用 OpsgenieNotifier 向 opsgenie 发送警报。

使用通知器¶

使用特定消息向 Opsgenie 发送警报。

tests/system/providers/opsgenie/example_opsgenie_notifier.py[源代码]

from __future__ import annotations

from datetime import datetime

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.providers.opsgenie.notifications.opsgenie import send_opsgenie_notification

with DAG(
    "opsgenie_notifier",
    start_date=datetime(2023, 1, 1),
    on_failure_callback=[send_opsgenie_notification(payload={"message": "Something went wrong!"})],
) as dag:
    BashOperator(
        task_id="mytask",
        bash_command="fail",
        on_failure_callback=[send_opsgenie_notification(payload={"message": "Something went wrong!"})],
    )