如何覆盖动态创建的DAG的start_date属性？

方案1：将DAG配置参数作为函数入参（推荐） #

这是最规范的实现方式，把需要灵活修改的DAG配置（比如start_date、dag_id、catchup）抽成some_dag_code的函数参数，在@dag装饰器中直接引用这些参数，调用时就能传入新值覆盖默认配置。

修改dag1.py：

import pendulum
from airflow.decorators import dag, task

@dag(
    dag_id=lambda kwargs: kwargs.get("dag_id", "my_dag"),
    start_date=lambda kwargs: kwargs.get("start_date", pendulum.datetime(2022, 9, 28, tz="UTC")),
    catchup=lambda kwargs: kwargs.get("catchup", True)
    # 其他需要自定义的DAG参数同理
)
def some_dag_code(name="bob", age=21, **dag_kwargs):
    @task()
    def print_name(name):
        print(name)
    
    @task()
    def print_age(age):
        print(age)
    
    task1 = print_name(name)
    task2 = print_age(age)

# 原DAG实例化
dag = some_dag_code(name="bob", age=21)

在dag2.py中调用：

import pendulum
from dags.dag1 import some_dag_code

dag = some_dag_code(
    name="michael", 
    age=31,
    dag_id="my_dag_but_for_michael",
    start_date=pendulum.datetime(2023, 3, 5, tz="UTC"),
    catchup=False
)

这种方式完全符合Airflow的设计逻辑，无兼容性风险，是生产环境的首选方案。

方案2：修改DAG的内部私有属性（不推荐） #

如果无法修改dag1.py的代码，可以尝试修改DAG对象的私有属性_start_date（注意：这是Airflow内部实现，不同版本可能变更，存在兼容性风险）：

import pendulum
from dags.dag1 import some_dag_code

dag = some_dag_code(name="michael", age=31)
dag.dag_id = "my_dag_but_for_michael"
# 修改私有属性实现覆盖
dag._start_date = pendulum.datetime(2023, 3, 5, tz="UTC")
dag.catchup = False

该方法能临时生效，但不建议在生产环境使用，Airflow版本更新可能导致代码失效。

方案3：重新初始化DAG对象（灵活但繁琐） #

提取原DAG的任务逻辑，重新创建新的DAG对象，完全自定义配置：

import pendulum
from airflow.decorators import dag
from dags.dag1 import some_dag_code

# 获取原DAG的任务定义
original_dag = some_dag_code(name="michael", age=31)

# 重新定义DAG，复用原任务
@dag(
    dag_id="my_dag_but_for_michael",
    start_date=pendulum.datetime(2023, 3, 5, tz="UTC"),
    catchup=False
)
def new_dag():
    # 复用原DAG中的任务并传入参数
    original_dag.tasks[0](name="michael")
    original_dag.tasks[1](age=31)

dag = new_dag()

这种方式灵活性极高，但需要手动处理任务的依赖和参数传递，适合需要大幅调整DAG配置的场景。

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