如何向Airflow内置模板传递操作符内的params参数？

问题场景 #

近期开始使用Airflow处理任务，学习模板功能时参考了EmailOperator的实现——它允许将HTML文件作为模板，访问ti.xcom_pull等内置变量。尝试实现类似的XML模板功能时，遇到问题：自定义的CreateXMLFromTemplate Operator中，希望XML模板文件能通过{{ params.xyz }}访问传入的params变量，但不清楚如何在渲染阶段传递该变量。当前代码如下：

class CreateXMLFromTemplate(BaseOperator):
    template_fields = ('xml_file')
    template_ext = ('.xml','.html')

    def __init__(
    self,
    *,
    xml_file,
    params,
    output_file,
    **kwargs,
    ):
        super().__init__(**kwargs)
        self.xml_file=xml_file
        self.params=params
        self.output_file=output_file
    
    def execute(self,context):
        with open(self.output_file,'w') as F:
            F.write(self.xml_file) #<- 已能渲染Airflow内置模板变量
        return self.output_file

解决方案 #

要让模板能访问自定义的params变量，需要将其注入Airflow的模板渲染上下文，具体修改如下：

1. 修正template_fields的定义 #

template_fields必须是元组类型，原代码中('xml_file')实际是字符串，会导致模板渲染异常，改为：

template_fields = ('xml_file',)

2. 重写get_template_context方法 #

通过重写Operator的get_template_context方法，将自定义的params变量合并到渲染上下文中，这样模板就能通过{{ params.xyz }}访问传入的参数：

class CreateXMLFromTemplate(BaseOperator):
    template_fields = ('xml_file',)
    template_ext = ('.xml','.html')

    def __init__(
        self,
        *,
        xml_file,
        params,
        output_file,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.xml_file = xml_file
        self.params = params
        self.output_file = output_file

    def get_template_context(self, context):
        # 获取默认的模板上下文
        template_context = super().get_template_context(context)
        # 将自定义params合并到上下文的params中（或直接替换，根据需求选择）
        if "params" in template_context:
            template_context["params"].update(self.params)
        else:
            template_context["params"] = self.params
        return template_context
    
    def execute(self,context):
        with open(self.output_file,'w') as F:
            F.write(self.xml_file)
        return self.output_file

原理说明 #

Airflow在渲染模板时，会调用Operator的get_template_context方法获取上下文变量。默认上下文包含ti（任务实例）、dag（DAG对象）等内置变量，我们通过重写该方法，将自定义的params加入上下文，模板就能直接通过{{ params.xyz }}引用对应参数。

额外说明 #

如果你的params需要和DAG/Task级别传入的params合并，上述代码中的update操作会保留原有params并添加自定义参数；如果希望完全使用Operator传入的params，直接替换template_context["params"] = self.params即可。

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