airflow使用branchpython运算符动态生成任务

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import BranchPythonOperator

def decide_which_path():
    # 在这里添加你的条件逻辑
    if condition:
        return 'task1', 'task1'
    else:
        return 'task2', 'task2'

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2021, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'dynamic_branch_example',
    default_args=default_args,
    description='A simple example of using BranchPythonOperator',
    schedule_interval=timedelta(days=1),
)

branching = BranchPythonOperator(
    task_id='branching',
    python_callable=decide_which_path,
    dag=dag,
)

task1 = DummyOperator(task_id='task1', dag=dag)
task2 = DummyOperator(task_id='task2', dag=dag)

branching >> [task1, task2]