Airflow 2.2.2中如何通过DAG参数设置http_conn_id？

首先明确：这种需求是可行的，不属于反模式，问题出在SimpleHttpOperator的http_conn_id字段默认不支持模板渲染，所以你写的{{ params.api }}会被当作字符串原样传入，不会解析成实际的连接ID。

解决方法有两种： #

1. 自定义支持模板的HttpOperator #

继承原SimpleHttpOperator，把http_conn_id加入模板字段列表：

from airflow.providers.http.operators.http import SimpleHttpOperator

class TemplatedSimpleHttpOperator(SimpleHttpOperator):
    # 扩展模板字段，添加http_conn_id
    template_fields = (*SimpleHttpOperator.template_fields, 'http_conn_id')

之后用这个自定义Operator替代原有的，http_conn_id='{{ params.api }}'就能正常解析DAG运行时传入的参数了。

2. 用PythonOperator结合HttpHook实现 #

如果不想自定义Operator，也可以用PythonOperator调用HttpHook，在Python函数里动态获取参数：

from airflow.providers.http.hooks.http import HttpHook
from airflow.operators.python import PythonOperator

def dynamic_api_call(**context):
    # 从上下文获取DAG传入的params参数
    target_conn_id = context["params"]["api"]
    # 初始化HttpHook
    http_hook = HttpHook(http_conn_id=target_conn_id, method="GET")
    # 执行请求，替换成你的实际端点
    response = http_hook.run(endpoint="/api/v1/data")
    # 这里可以添加响应处理逻辑，比如打印、写入数据库等
    print(f"Response status: {response.status_code}")

# 定义PythonOperator任务
dynamic_api_task = PythonOperator(
    task_id="dynamic_api_call",
    python_callable=dynamic_api_call,
    provide_context=True,
    # 运行DAG时可以传入params覆盖这里的默认值
    params={"api": "default_api_conn"}
)

补充说明 #

参数化连接ID在多环境切换、动态选择不同API源等场景下非常实用，完全是合理的用法，不属于反模式。只要确保传入的params.api对应的连接ID在Airflow的连接管理中已经配置即可。

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