Airflow自定义算子如何同时保留原始文件名与渲染后的文件内容？

问题核心在于：Airflow对序列类型（如列表）的模板字段会进行递归渲染，且Python中列表是可变对象——你在__init__中直接赋值self.inputs_sql_original = inputs_sql时，两个变量指向同一个列表对象。当Airflow渲染inputs_sql（标记为template_fields）时，会直接修改列表内的元素，导致原始文件名被覆盖。而字符串是不可变对象，赋值后不会被共享修改，这就是字符串类型能保留原始值的原因。

解决方案：复制原始列表，切断对象引用 #

在__init__方法中对输入的文件名列表进行拷贝，让inputs_sql_original指向独立的列表对象，避免被Airflow的渲染逻辑修改。

修改后的代码示例：

from typing import Sequence
from airflow import DAG
from airflow.models import BaseOperator
from pendulum import datetime

class RenderingOperator(BaseOperator):

    template_fields: Sequence[str] = ("inputs_sql",)
    template_ext: Sequence[str] = (".sql",)

    def __init__(self, inputs_sql, **kwargs):
        super().__init__(**kwargs)
        # 拷贝列表，创建独立的原始文件名列表
        self.inputs_sql_original = inputs_sql.copy()  # 或使用 list(inputs_sql)
        self.inputs_sql = inputs_sql

    def execute(self, context):
        print("self.inputs_sql_original (原始文件名):")
        print(self.inputs_sql_original)
        print("self.inputs_sql (渲染后内容):")
        print(self.inputs_sql)

with DAG(
    dag_id="check_rendering",
    start_date=datetime(2024, 1, 1),
    schedule=None
) as dag:
    RenderingOperator(
        task_id="render_sql",
        inputs_sql=["query/query_0.sql"],
        dag=dag,
    )

预期输出 #

self.inputs_sql_original (原始文件名):
['query/query_0.sql']
self.inputs_sql (渲染后内容):
['select * from table\n']

补充说明 #

• 若列表中包含嵌套可变对象（如子列表、字典），需使用copy.deepcopy(inputs_sql)进行深拷贝，确保完全切断引用；但本例中仅为字符串元素，浅拷贝copy()或list()已足够。
• 确保inputs_sql_original未被加入template_fields，否则Airflow仍会对其进行渲染处理。

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