Airflow Operator开发：如何根据Run ID修改Task Context？

你尝试直接修改context['dag']的回调方法无效，核心原因是Airflow在任务执行阶段，DAG对象的回调配置已经固化，运行时修改不会被调度器识别。以下是几个可行的解决方案：

方案一：在execute方法中手动触发回调 #

这是最直接可靠的方式，直接在任务执行完成后根据run_id判断是否调用回调函数：

class MyOperator(BaseOperator):
    def __init__(self, ..., **kwargs):
        super().__init__(**kwargs)
        ...

    def execute(self, context):
        dr: DagRun = context["dag_run"]
        run_id = dr.run_id
        is_manual = run_id.startswith('manual__')
        
        # 执行你的核心任务逻辑
        ...
        
        # 任务成功后手动触发回调
        if is_manual:
            my_alert_function(context)

如果需要处理失败场景，可通过try-except块捕获异常，在异常分支调用失败回调：

def execute(self, context):
    dr: DagRun = context["dag_run"]
    is_manual = dr.run_id.startswith('manual__')
    try:
        # 核心任务逻辑
        ...
        # 成功回调
        if is_manual:
            my_alert_function(context)
    except Exception as e:
        # 失败回调
        if is_manual:
            my_alert_function(context)
        raise e  # 重新抛出异常，保留Airflow的任务失败状态记录

方案二：动态修改当前Task实例的回调 #

若希望沿用Airflow原生的回调触发机制，可尝试修改当前任务实例的回调属性（部分Airflow版本可能存在兼容性问题，建议测试验证）：

def execute(self, context):
    dr: DagRun = context["dag_run"]
    is_manual = dr.run_id.startswith('manual__')
    task = context["task"]
    
    if is_manual:
        # 修改当前任务的成功/失败回调
        task.on_success_callback = my_alert_function
        task.on_failure_callback = my_alert_function
    
    # 执行核心任务逻辑
    ...

方案三：在DAG初始化时配置动态回调（扩展思路） #

如果你的Operator固定服务于特定DAG，可在DAG定义阶段设置一个动态判断的回调函数，由它根据run_id决定是否触发目标逻辑：

def dynamic_callback(context):
    run_id = context["dag_run"].run_id
    if run_id.startswith('manual__'):
        my_alert_function(context)

dag = DAG(
    dag_id="target_dag",
    on_success_callback=dynamic_callback,
    on_failure_callback=dynamic_callback,
    ...
)

with dag:
    MyOperator(task_id="my_task")

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