airflow_dag_opsgenie_create_alert_operator_example.py

1import datetime
2
3from airflow import DAG
4from airflow.providers.opsgenie.operators.opsgenie import OpsgenieCreateAlertOperator
5
6with DAG(
7    dag_id="example_opsgenie_alert",
8    schedule=None,
9    start_date=datetime.datetime(2021, 1, 1),
10    catchup=False,
11    tags=["example"],
12) as dag:
13
14    # [START howto_operator_opsgenie_create_alert]
15    create_alert = OpsgenieCreateAlertOperator(
16        task_id="create_alert_task",
17        message="An example alert message",
18        alias="example_alias",
19        description="This is a detailed description of the alert",
20        responders=[{"id": "6139f35d-206d-4071-8072-50d605206263", "type": "team"}],
21        visible_to=[
22            {"id": "6139f35d-206d-4071-8072-50d605206263", "type": "team"},
23            {"username": "testuser@test.com", "type": "user"},
24        ],
25        actions=["Action1", "Action2"],
26        tags=["Tag1", "Tag2"],
27        details={"key1": "value1", "key2": "value2"},
28        entity="An example entity",
29        priority="P1",
30        opsgenie_conn_id="opsgenie_default",
31    )
32    # [END howto_operator_opsgenie_create_alert]