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


def pagerduty_on_failure_callback(context):
    hook = PagerdutyHook(pagerduty_conn_id=context["params"]["pagerduty_conn_id"])
    dag = context["dag"]
    task = context["task"]
    run_id = context["run_id"]
    hook.create_event(
        summary=f"Execution failure in DAG {dag.dag_id} run {run_id} at task {task.task_id}.",
        severity="error",
        source=dag.dag_id,
    )
