自定义Airflow传感器中Jinja模板的异常行为问题

Airflow的模板渲染逻辑是运行时触发，而非DAG解析阶段。你的代码存在两个核心问题：

1. __init__方法在DAG加载时就执行了，此时传入的{{ ts }}只是原始字符串，还没被Airflow渲染。
2. self.time不属于Airflow的模板字段列表，所以Airflow不会自动对它进行渲染，导致它始终保留原始的模板字符串。

而bash_command能正确渲染，是因为它是BashSensor内置的template_fields之一，Airflow会在任务运行的poke阶段自动处理它的模板替换。

修改自定义传感器，将time加入模板字段列表，让Airflow自动渲染该字段：

class MySensor(BashSensor):
    # 继承BashSensor的模板字段，追加自定义的time字段
    template_fields = BashSensor.template_fields + ('time',)

    def __init__(self, time, **kwargs):
        self.time = time
        cmd = f"java some-other-stuff {self.time}"
        super().__init__(**kwargs, bash_command=cmd)

    def poke(self, context):
        # 必须传入context，确保父类能完成模板渲染
        status = super().poke(context)
        if status:
            print(self.time)  # 此时self.time已被Airflow渲染为实际时间戳
        else:
            print("trying again")
        return status

如果你的场景只需要获取ts，也可以直接从poke方法的context参数中提取，无需依赖模板字段：

def poke(self, context):
    status = super().poke(context)
    if status:
        # 直接从context中拿渲染后的ts
        print(context['ts'])
    else:
        print("trying again")
    return status

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