Airflow -2警报已发送on_failure

[smtp]
smtp_host = smtp.example.com
smtp_starttls = True
smtp_ssl = False
smtp_user = your_email@example.com
smtp_password = your_password
smtp_port = 587
smtp_mail_from = your_email@example.com

from airflow.utils.email import send_email
from airflow.hooks.base_hook import BaseHook

def task_failure_alert(context):
    dag_id = context.get('dag').dag_id
    task_id = context.get('task').task_id
    execution_date = context.get('execution_date')
    log_url = context.get('task_instance').log_url

    subject = f"Airflow alert: {dag_id}.{task_id} Failed"
    html_content = f"""
    <h3>Task Failed</h3>
    <p><strong>Dag:</strong> {dag_id}</p>
    <p><strong>Task:</strong> {task_id}</p>
    <p><strong>Execution Time:</strong> {execution_date}</p>
    <p><strong>Log URL:</strong> <a href="{log_url}">{log_url}</a></p>
    """

    send_email('alert@example.com', subject, html_content)

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    'on_failure_callback': task_failure_alert
}

dag = DAG(
    'example_failure_alert',
    default_args=default_args,
    description='A simple tutorial DAG',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2023, 1, 1),
    catchup=False,
)

def failing_task():
    raise ValueError("This task is supposed to fail.")

start = DummyOperator(
    task_id='start',
    dag=dag,
)

fail = PythonOperator(
    task_id='fail',
    python_callable=failing_task,
    dag=dag,
)

start >> fail