Airflow技术问询：如何通过dag_run配置动态设置PythonOperator的trigger_rule

由于trigger_rule不属于Airflow的模板化字段，无法直接通过Jinja表达式从dag_run.conf取值，你可以通过自定义Operator子类的方式实现需求，具体步骤如下：

1. 自定义继承自PythonOperator的新Operator，重写get_trigger_rule方法，让它从DAG运行配置中读取触发规则：

from airflow.operators.python import PythonOperator

class DynamicTriggerRulePythonOperator(PythonOperator):
    def get_trigger_rule(self):
        # 优先从当前DAG运行的配置中获取trigger_rule，无配置则用默认值
        if self.dag_run and self.dag_run.conf:
            return self.dag_run.conf.get('trigger_rule', super().get_trigger_rule())
        # 无dag_run上下文时（如DAG解析阶段）返回Operator定义时的默认值
        return super().get_trigger_rule()

2. 在DAG中使用这个自定义Operator替代原生PythonOperator：

from datetime import datetime
from airflow import DAG

def my_func():
    # 你的任务业务逻辑
    print("执行任务")

with DAG(
    dag_id='dynamic_trigger_dag',
    schedule_interval=None,
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    test_task = DynamicTriggerRulePythonOperator(
        task_id='test',
        python_callable=my_func,
        trigger_rule='all_success'  # 这里的默认值会被触发时的配置覆盖
    )

3. 触发DAG时，在配置参数中传入{"trigger_rule": "all_done"}，任务就会使用该触发规则执行。

这种方法的核心是利用Airflow Operator的get_trigger_rule方法在运行时动态返回规则值，完美适配你需要从触发配置中动态获取的场景。

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