Airflow中开启provide_context时op_kwargs参数会传入ds还是**kwargs？

结论 #

部署到Airflow运行后，你传入的client_production会进入**kwargs参数，不会出现在ds参数中。

原理说明 #

• 当PythonOperator开启provide_context=True配置时，Airflow调用目标函数的规则是固定的：第一个位置参数会自动传入执行日期的YYYY-MM-DD格式字符串，对应你函数定义里的ds参数，该参数不会接收任何你自定义的op_kwargs值。
• 所有在op_kwargs中配置的自定义参数，都会以关键字参数的形式注入到**kwargs中，你需要通过kwargs['client']的方式读取client_production变量。

本地测试错误原因 #

你本地的调用方式不符合Airflow的实际调用逻辑：直接把字典作为第一个位置参数传入函数，自然会被ds参数接收。如果要本地模拟Airflow的调用逻辑，应该用如下方式测试：

# 第一个参数传入模拟的执行日期字符串，自定义参数以关键字参数形式传入
get_campaign_active("2024-01-01", client=client_production)

此时client_production就会正常进入**kwargs中，和线上运行表现一致。

代码修正建议 #

你当前函数里读取client的逻辑是错误的，需要调整为从kwargs读取：

def get_campaign_active(ds, **kwargs):
    logging.info('Checking for inactive campaign types..')
    # 从kwargs中读取client参数
    the_db = kwargs['client']
    db = the_db['misc-server']
    collection = db.campaigntypes
    campaign = list(collection.find({})) 
    for item in campaign:
        if item['active'] == False:
            # storing false 'active' campaigns
            result = "'{}' active status set to False".format(item['text'])
            logging.info("'{}' active status set to False".format(item['text']))

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