如何在Airflow失败Slack告警中添加对应DAG的页面URL？

在Airflow 2.2.1中，dag_code并不是context内置变量，无法通过它获取DAG的URL。正确做法是通过Airflow配置获取Webserver基础地址，结合DAG ID拼接目标URL，具体实现如下：

实现步骤 #

• 从Airflow配置读取Webserver基础URL（推荐方式，避免硬编码）；
• 用已获取的dag_id拼接出DAG代码页面的完整URL；
• 将URL格式化为Slack支持的链接样式，加入告警消息。

修改后的代码 #

更新alert_slack函数 #

from airflow.providers.slack.operators.slack import SlackAPIPostOperator
from airflow.configuration import conf  # 导入配置模块

def alert_slack(channel: str):
    def failure(context):
        last_task = context.get('task_instance')
        task_name = last_task.task_id
        dag_name = last_task.dag_id
        # 获取Airflow Webserver基础URL
        airflow_base_url = conf.get('webserver', 'base_url')
        # 拼接DAG代码页面URL并格式化为Slack可识别的链接
        dag_url = f"<{airflow_base_url}/code?dag_id={dag_name}|DAG 代码页面>"
        log_link = f"<{last_task.log_url}|{task_name}>"
        error_message = context.get('exception') or context.get('reason')
        execution_date = context.get('execution_date')
        # 修复owner变量未定义问题，从DAG对象中获取
        owner = last_task.dag.owner
        title = f':red_circle: DAG Failed.'
      
        msg_parts = {
            '*Dag*': dag_name,
            '*Owner*': owner,
            '*Task*': task_name,
            '*Log*': log_link,
            '*DAG 页面*': dag_url,  # 添加新的DAG URL字段
            '*Error*': error_message,
            '*Execution date*': execution_date
        }
        msg = "\n".join([title,
            *[f"{key}: {value}" for key, value in msg_parts.items()]
        ]).strip()
        SlackAPIPostOperator(
            task_id="alert",
            slack_conn_id="slack_alert",
            text=msg,
            channel=channel,
        ).execute(context=None)
    return failure

补充说明 #

• 若Airflow Webserver基础URL未在配置中设置，可临时硬编码（不推荐，不利于环境迁移），例如：airflow_base_url = "https://myairflow.dev"；
• 如需添加DAG监控页面而非代码页面，只需修改路径为/dags/{dag_name}，即dag_url = f"<{airflow_base_url}/dags/{dag_name}|DAG 监控页面>"。

内容的提问来源于stack exchange，提问作者Zaoza14