Airflow 2.9.0中Python Operator报op_kwargs参数重复TypeError

错误原因 #

你遇到的TypeError: got multiple values for keyword argument 'op_kwargs'，核心原因是在DAG上下文管理器内部定义任务时，给@task.sensor装饰器额外传递了dag=dag参数。

在Airflow中，当你处于with DAG(...) as dag:的代码块内时，所有通过装饰器定义的任务会自动关联到当前DAG，无需显式指定dag=dag。而Airflow 2.9.0的装饰器内部逻辑会把你显式传递的dag参数错误地合并到op_kwargs中，导致Operator初始化时收到重复的op_kwargs参数。

修复步骤 #

直接删除@task.sensor装饰器中的dag=dag参数即可，修改后的代码如下：

@task.sensor(
    task_id='some_task_op',
    mode='reschedule',  # default is 'poke'
    poke_interval=10*60,
    timeout=4*60*60,
    op_kwargs={
        'creds': '{{ var.value.get("api-credentials") }}',
    },
)
def some_task_op(params, data_interval_end, creds=None):
    pass

额外说明 #

• 函数参数params和data_interval_end是Airflow自动注入的上下文变量，不需要在op_kwargs中手动传递，当前写法没问题。
• 只有在DAG上下文外部需要关联任务到指定DAG时，才需要显式传递dag=dag，这种场景极少。

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