Airflow自定义Operator的Jinja模板为何仅部分生效？

我按照Airflow官方文档所述，使用Jinja模板对自定义Operator进行参数化，编写代码如下：

class HelloOperator(BaseOperator):

    template_fields: Sequence[str] = ("name",)

    def f(self, name):
        'This function does nothing'
        return name

    def __init__(self, name: str, **kwargs) -> None:
        super().__init__(**kwargs)
        self.name = name
        self.name2 = self.f(name)
        self.name3 = self.f(self.name)

    def execute(self, context):
        message = f"Hello from {self.name}"
        print(message)
        message2 = f"Hello from {self.name2}"
        print(message2)
        message3 = f"Hello from {self.name3}"
        print(message3)
        return message

执行后得到如下输出：

[2022-12-07, 17:33:23 UTC] {logging_mixin.py:115} INFO - Hello from task_id_1
[2022-12-07, 17:33:23 UTC] {logging_mixin.py:115} INFO - Hello from {{ task_instance.task_id }}
[2022-12-07, 17:33:23 UTC] {logging_mixin.py:115} INFO - Hello from {{ task_instance.task_id }}

可见仅self.name完成了模板渲染，name2和name3仍保留Jinja表达式。请问为何模板仅在第一种情况生效？文档中提到“Jinja替换的是Operator属性而非参数”，该说明如何适用于第二、第三种情况？

1. 模板仅对self.name生效的原因 #

Airflow的Jinja模板渲染流程是在Operator的__init__方法执行完毕之后，execute方法执行之前触发的。只有被声明在template_fields列表里的实例属性，才会被Airflow自动进行模板替换。

你在__init__中计算name2和name3时，模板还未开始渲染：

• self.name2 = self.f(name)：这里的name是传入构造函数的原始Jinja表达式字符串，还没被替换；
• self.name3 = self.f(self.name)：此时self.name刚被赋值为原始的Jinja字符串，同样未经过渲染。

后续Airflow只会处理template_fields里的name属性，而name2、name3不在该列表中，所以它们的内容会保持未渲染的状态直到执行阶段。

2. 对“Jinja替换的是Operator属性而非参数”的理解 #

这里的“参数”指的是你传给Operator构造函数的入参（比如__init__里的name参数），Airflow不会直接修改这些入参的值；而“属性”指的是Operator实例的属性（比如self.name），Airflow只会对template_fields中指定的实例属性做模板替换。

但关键是替换时机在__init__之后：你在__init__里用未渲染的入参或属性计算得到的name2、name3，本质是把原始Jinja字符串赋值给了这两个属性，且它们不在template_fields中，所以Airflow不会对它们进行后续替换。

3. 解决方案 #

最合理的方式是把name2、name3的计算逻辑移到execute方法中，此时self.name已经完成模板渲染：

class HelloOperator(BaseOperator):

    template_fields: Sequence[str] = ("name",)

    def f(self, name):
        'This function does nothing'
        return name

    def __init__(self, name: str, **kwargs) -> None:
        super().__init__(**kwargs)
        self.name = name

    def execute(self, context):
        # 此时self.name已完成模板渲染
        self.name2 = self.f(self.name)
        self.name3 = self.f(self.name)
        
        message = f"Hello from {self.name}"
        print(message)
        message2 = f"Hello from {self.name2}"
        print(message2)
        message3 = f"Hello from {self.name3}"
        print(message3)
        return message

执行后，name2和name3会基于渲染后的self.name生成，输出结果会全部显示为Hello from task_id_1。

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