如何使用Taskflow API向任务传递Jinja2模板化参数

问题根因 #

你遇到的问题来自Airflow Jinja渲染的两个核心特性：

1. @task 装饰器默认不会自动渲染函数的参数默认值，只会渲染你调用任务时传入的参数。如果要让默认值也触发渲染，需要显式在装饰器中声明template_fields列表，传入需要渲染的参数名。
2. 你代码中hmac_secret_key的默认值后直接加了.encode('utf-8')，这行代码会在DAG解析阶段（也就是函数定义时）立即执行，根本等不到任务运行时的模板渲染步骤，你之前感知到它能正常工作大概率是你调用任务时主动传了该参数的值，没有走默认值。

解决方案 #

两种方案都可以实现运行时渲染模板，不会在DAG解析阶段连接元数据库：

方案1：保留参数默认值写法 #

给@task装饰器添加template_fields指定要渲染的参数，同时把字符串编码逻辑移到函数内部，避免DAG解析阶段提前执行：

from airflow.decorators import task

@task(
    template_fields=["api_key", "hmac_secret_key"] # 显式声明需要渲染的参数
)
def generate_signature_headers(
    api_key="{{ var.json.knmi_api_key.api_key }}",
    hmac_secret_key="{{ var.json.knmi_api_key.hmac_secret_key }}" # 去掉默认值后的encode，移到函数内
):
    # 编码操作移到函数内部，此时参数已经完成Jinja渲染
    hmac_secret_key_bytes = hmac_secret_key.encode('utf-8')
    hmac_digest_transformed = ...原有业务逻辑...

    return {
        "Date": now_utc,
        "Authorization": f'Signature keyId="{api_key}",algorithm="hmac-sha512",'
        f'signature="{hmac_digest_transformed}" ',
    }

方案2：调用任务时传递模板参数 #

如果不想修改函数默认值，也可以在DAG中调用该任务时，把模板作为参数传入，TaskFlow默认会自动渲染调用时传入的模板值，不需要额外声明template_fields：

# 函数定义简化，不需要写默认值也不需要加template_fields
@task
def generate_signature_headers(api_key, hmac_secret_key):
    hmac_secret_key_bytes = hmac_secret_key.encode('utf-8')
    # 其余逻辑不变

# 在DAG的任务流中调用时传入模板
with DAG(...) as dag:
    generate_signature_headers(
        api_key="{{ var.json.knmi_api_key.api_key }}",
        hmac_secret_key="{{ var.json.knmi_api_key.hmac_secret_key }}"
    )

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