如何对非Airflow变量强制启用Jinja模板渲染？

你问的这个问题在Airflow开发里挺常见的——咱们都知道Airflow自带的那些标记为templated的参数能直接用Jinja渲染，像PythonOperator的部分参数也会自动支持，但碰到自己定义的变量、非Airflow原生的参数时，就得用一些专门的方式来处理了。我结合你提到的PythonOperator场景，给你拆解几种靠谱的实现方式：

1. 用PythonOperator的templates_dict传递模板化参数 #

这是最常用的方案，Airflow专门为PythonOperator提供了templates_dict参数，它本身就支持Jinja模板渲染。你可以把需要模板化的自定义变量都放在这里，然后在Python函数里通过上下文（context）来获取渲染后的值。

示例代码：

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

def my_python_func(**context):
    # 从context里取出templates_dict中渲染后的自定义变量
    custom_rendered_val = context['templates_dict']['my_custom_var']
    print(f"渲染后的自定义变量: {custom_rendered_val}")
    # 也能直接通过context获取Airflow内置变量，比如执行日期
    exec_date = context['execution_date']
    print(f"当前任务执行日期: {exec_date.strftime('%Y-%m-%d')}")

with DAG(
    dag_id='custom_template_demo',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily',
    catchup=False
) as dag:
    template_task = PythonOperator(
        task_id='render_custom_vars',
        python_callable=my_python_func,
        # 这里的templates_dict支持任意Jinja模板语法
        templates_dict={
            'my_custom_var': "今天是 {{ execution_date.strftime('%Y-%m-%d') }}，任务ID是 {{ ti.task_id }}",
            'from_airflow_var': "{{ var.value.my_custom_global_var }}"
        },
        # 必须开启这个参数，才能让函数拿到包含templates_dict的上下文
        provide_context=True
    )

2. 手动调用Airflow模板环境渲染 #

如果你的场景更灵活（比如想在函数内部动态渲染模板，或者变量不在Operator参数里），可以直接调用Airflow的Jinja模板环境来手动渲染字符串。

示例代码：

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

# 获取Airflow内置的Jinja模板环境
template_env = JinjaTemplate().environment

# 自定义需要模板化的字符串
custom_template_str = "执行日期: {{ ds }}，我的自定义变量值是 {{ var.value.test_variable }}"

def manual_render_func(**context):
    # 传入上下文（包含Airflow所有内置变量）进行渲染
    rendered_result = template_env.from_string(custom_template_str).render(context)
    print(f"手动渲染后的结果: {rendered_result}")

with DAG(
    dag_id='manual_template_demo',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily',
    catchup=False
) as dag:
    manual_task = PythonOperator(
        task_id='manual_render_task',
        python_callable=manual_render_func,
        provide_context=True
    )

3. 给自定义Operator的参数标记为templated #

如果你是自己开发自定义Operator，想要让某个参数支持模板化，只需要在Operator类里定义templated_fields属性，把需要模板化的参数名加进去就行——Airflow会自动帮你完成渲染。

示例代码：

from airflow.models.baseoperator import BaseOperator
from airflow.utils.decorators import apply_defaults

class MyCustomOperator(BaseOperator):
    # 标记需要支持模板化的参数
    templated_fields = ('custom_param',)

    @apply_defaults
    def __init__(self, custom_param, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.custom_param = custom_param

    def execute(self, context):
        # 这里的self.custom_param已经是Airflow渲染后的结果了
        print(f"自定义参数渲染后的值: {self.custom_param}")

# 在DAG中使用自定义Operator时，直接传带Jinja语法的字符串即可
with DAG(
    dag_id='custom_operator_template_demo',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily',
    catchup=False
) as dag:
    custom_op_task = MyCustomOperator(
        task_id='custom_op_template_task',
        custom_param="当前任务执行时间: {{ execution_date }}"
    )

几个关键注意点 #

• 不管用哪种方式，都要确保你需要的Airflow内置变量（比如execution_date、var、ds、ti等）能被模板环境访问到，通常通过provide_context=True或者在Operator的execute方法里获取上下文来传递。
• 尽量避免在DAG定义阶段直接渲染模板（除非你要的是静态值），因为DAG会被定时解析，这时execution_date这类动态变量还没有具体值，会导致渲染错误。

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