如何在Python编写的Airflow DAG中获取{{ ds }}等内置变量的值

原生Python代码无法直接通过my_date = '{{ ds }}'的写法获取到Airflow内置变量值，你遇到的不生效问题是因为Jinja模板渲染仅发生在任务（Operator）的执行阶段，DAG文件本身的解析加载阶段不会做模板替换。

要实现Python代码和SQL的变量取值对齐，也就是拿到和{{ ds }}完全一致、适配回填场景的执行日期，按照使用场景可以参考以下方案：

1. 在PythonOperator中使用 #

这是最常见的场景，直接从任务传入的上下文（context）中取值即可：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def your_business_func(**context):
    # 取值和SQL里的{{ ds }}完全一致，回填时自动对应历史执行日期
    current_exec_date = context["ds"]
    # 也可以取其他内置模板变量，对应名称和Jinja模板里的变量名完全相同
    current_exec_date_nodash = context["ds_nodash"]
    prev_exec_date = context["prev_ds"]
    # 后续业务逻辑全部使用上述变量即可
    print(f"当前执行对应日期：{current_exec_date}")

with DAG(
    dag_id="sample_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
) as dag:
    sample_task = PythonOperator(
        task_id="sample_task",
        python_callable=your_business_func,
        # Airflow 2.x 版本默认开启provide_context，无需显式声明，1.x版本需要加这行
        provide_context=True
    )

2. 在其他Operator中使用动态变量 #

如果需要在非PythonOperator的参数中使用经过Python逻辑处理的变量，可以将逻辑封装为支持Jinja模板的形式，或者自定义Operator的可模板化字段，最终变量取值依然从执行上下文获取即可。

注意：绝对不要使用date.today()、datetime.now()这类获取系统当前时间的方法，这类值只和代码运行时刻的系统时间有关，和Airflow的执行周期、回填的历史日期完全无关，无法和SQL逻辑对齐。

所有Airflow内置的Jinja模板变量，都可以在任务执行阶段的上下文中拿到，变量名完全对应，取值和SQL场景下的模板渲染结果完全一致，可以满足对齐要求。

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