Airflow如何为dag_run.conf设置默认值避免重复定义参数

只有通过Web UI手动携带配置触发DAG时，dag_run.conf才会包含传入的参数。定时调度触发的DAG，dag_run.conf默认是空字典，直接读取不存在的key就会抛出jinja2.exceptions.UndefinedError报错。

最优实现（无需重复定义默认值） #

直接使用Airflow内置的Jinja上下文变量params替代dag_run.conf即可，Airflow原生支持参数自动合并逻辑：

• 定时运行无自定义配置时，自动读取DAG/Operator层级定义的params默认值
• 手动通过「Run w/ Config」传入自定义配置时，会自动覆盖对应key的默认值，未覆盖的key仍然使用默认值

修改后的代码如下：

from airflow import DAG
from airflow.utils.dates import hours_ago
from airflow.operators.bash import BashOperator
from datetime import timedelta

def do_something(val1: str, val2: str) -> str:
    return f'echo vars are: "{val1}, {val2}"'

params = {
    'key1': 'def1',
    'key2': 'def2',        
}

default_args = {
    'retries': 0,
}

with DAG(
    'template_test',
    default_args=default_args,
    schedule_interval=timedelta(minutes=1),
    start_date=hours_ago(1),
    params = params,
) as dag:

    hello_t = BashOperator(
        task_id='example-command',
        # 将dag_run.conf替换为内置params变量
        bash_command=do_something('{{ params["key1"] }}', '{{ params["key2"] }}'),
    )

兼容方案（必须使用dag_run.conf的场景） #

如果业务逻辑必须读取dag_run.conf的内容，可通过Jinja的字典默认取值方法实现兜底，不需要额外写if/else判断：

{{ dag_run.conf.get('key1', params.key1) }}

或者更简洁的简写形式：

{{ (dag_run.conf or params).key1 }}

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