Dependencies & Flow Control

29Declaring edges#

What it does: Wires task dependencies using >> (downstream), << (upstream), or the explicit set_upstream()/set_downstream() methods.

How it works:

• a >> b >> c chains left to right (b starts after a; c starts after b). a << b << c reads "a depends on b, which depends on c," producing c -> b -> a.
• Fan-out/fan-in with a list on one side works cleanly: a >> [b, c] (one-to-many) and [b, c] >> d (many-to-one) are both fully supported.
• A list on both sides ([a, b] >> [c, d]) is not supported — expand it into explicit pairs instead (see the example).
• These calls only build the dependency graph — the actual ready/blocked/skip decision for each task is governed by its trigger_rule (feature 14).

Example:

with DAG(dag_id="edges_demo", schedule="@daily", start_date=datetime(2026, 1, 1)) as dag:
    extract = PythonOperator(task_id="extract", python_callable=lambda: None)
    validate = PythonOperator(task_id="validate", python_callable=lambda: None)
    transform = PythonOperator(task_id="transform", python_callable=lambda: None)
    load_a = PythonOperator(task_id="load_a", python_callable=lambda: None)
    load_b = PythonOperator(task_id="load_b", python_callable=lambda: None)
    report = PythonOperator(task_id="report", python_callable=lambda: None)

    # Simple chain
    extract >> validate >> transform

    # Fan-out: one task to many (list on ONE side)
    transform >> [load_a, load_b]

    # Fan-in: many tasks to one (list on ONE side)
    [load_a, load_b] >> report

    # A list on BOTH sides isn't supported. Expand it explicitly instead:
    # for src in [load_a, load_b]:
    #     for dst in [report, some_other_task]:
    #         src >> dst

30Edge labels#

What it does: Annotates an edge with a short text label for the task-graph visualization — purely cosmetic, no effect on execution.

Parameters:

How it works:

• Chain it directly into a dependency declaration: task_a >> Label("on_success") >> task_b. The explicit method form, task_a.set_downstream(task_b, label="on_success"), produces the same result if you prefer it.
• Labels only affect what's displayed in the task-graph view; they never influence trigger-rule evaluation, scheduling, or execution order.

Example:

from dag_parser.dynamic.dag_context import Label

with DAG(dag_id="edge_labels_demo", schedule="@daily", start_date=datetime(2026, 1, 1)) as dag:
    check = PythonOperator(task_id="check", python_callable=lambda: None)
    proceed = PythonOperator(task_id="proceed", python_callable=lambda: None)

    check >> Label("on_success") >> proceed

31Convergence control#

What it does: Explains how a "join" task — one with multiple upstream edges — decides whether it's ready, using its trigger_rule (feature 14).

How it works:

• Declaring edges (feature 29) only builds the graph — it never implies join semantics on its own. A task with 3 upstream edges and the default trigger_rule="all_success" requires all 3 to succeed; use one_success (or another applicable rule) if any one finishing is enough.
• The classic pattern: after a BranchPythonOperator skips every path but one, a naive all_success join downstream of all branches would never be satisfied (skipped branches aren't "success"). Use none_failed or none_failed_min_one_success on the join so the unchosen branches' skip state doesn't permanently block it.
• A converging task with an all_success/all_done-style rule effectively waits for every incoming edge's task to reach a terminal state, even if most of them finished quickly and one is slow.
• Mixing trigger rules across parallel join tasks fed by the same upstream fan-out is fine and common — e.g. one join using one_success for a "fast path" notification, and another using all_success as the real completion gate.

Example:

with DAG(dag_id="convergence_demo", schedule="@daily", start_date=datetime(2026, 1, 1)) as dag:

    choose_path = PythonOperator(task_id="choose_path", python_callable=lambda: None)
    path_a = PythonOperator(task_id="path_a", python_callable=lambda: None)
    path_b = PythonOperator(task_id="path_b", python_callable=lambda: None)

    # Converges after a branch — must tolerate the unchosen path being 'skipped'
    join_after_branch = PythonOperator(
        task_id="join_after_branch",
        python_callable=lambda: None,
        trigger_rule="none_failed_min_one_success",
    )

    choose_path >> [path_a, path_b] >> join_after_branch

32Dynamic task mapping#

What it does: Fans a single task definition out into N task instances at runtime, one per element of an iterable — a literal list or an upstream task's XCom output.

Parameters:

How it works:

• Passing a single keyword argument to .expand() produces one mapped instance per element, in order.
• Passing multiple keyword arguments produces the cross-product of every combination — .expand(a=[1, 2], b=["x", "y"]) produces 4 instances: (1,"x"), (1,"y"), (2,"x"), (2,"y").
• If the resolved iterable has zero elements, the whole mapped task is skipped (not "zero instances silently"). Downstream joins should use none_failed/none_failed_min_one_success (features 14/31) so they aren't permanently blocked by that.
• When expanding off an upstream task's XCom (XComArg), the referenced return_value can be either a JSON array (one element per instance) or a single scalar value, which is automatically treated as a one-element list.
• .partial(**fixed_kwargs) values are merged into every expanded instance first; each expand key's value is merged on top per instance.
• Retries, timeouts, pool, and trigger rule come from the shared task definition — every mapped instance shares the same settings; only the expanded arguments differ per instance.

Example:

from dag_parser.dynamic.dag_context import XComArg

with DAG(dag_id="fanout_files", schedule="@daily", start_date=datetime(2026, 1, 1)) as dag:

    def list_files():
        return ["file_a.csv", "file_b.csv", "file_c.csv"]

    list_task = PythonOperator(task_id="list_files", python_callable=list_files)

    def process_file(filename, region):
        print(f"processing {filename} in {region}")

    # Two expand keys -> cross-product of filenames x regions
    process = PythonOperator.partial(
        task_id="process_file",
        python_callable=process_file,
    ).expand(
        filename=XComArg("list_files"),   # or a literal list: filename=["a.csv", "b.csv"]
        region=["us-east", "us-west"],
    )

    list_task >> process

33Branching#

What it does: BranchPythonOperator runs a function whose return value selects which of its direct downstream tasks actually run; every other direct downstream task is marked skipped.

Parameters:

How it works:

• The callable must always return something — a single task_id or a list of them. Forgetting to return (or returning None) fails the task outright, rather than skipping everything.
• A returned task_id that isn't actually a direct downstream of the branch task is simply ignored (logged, not an error) — that path just never gets selected.
• Skip-cascading downstream of an unselected branch is trigger-rule aware — a join using none_failed_min_one_success fed by both the chosen and unchosen branch is correctly left alone (not skipped) as long as the chosen branch can still succeed. Tasks with trigger_rule="always" are never cascade-skipped.
• The branch decision is pushed as the task's normal return_value XCom — you can also read it yourself downstream via xcom_pull if useful.

Example:

from dag_parser.dynamic.operators import BranchPythonOperator

with DAG(dag_id="branch_demo", schedule="@daily", start_date=datetime(2026, 1, 1)) as dag:

    def choose_branch(**context):
        if context["ds"] < "2026-06-01":
            return "legacy_path"          # single task_id
        return ["new_path_a", "new_path_b"]  # or a list of task_ids

    branch = BranchPythonOperator(
        task_id="choose_branch",
        python_callable=choose_branch,
        provide_context=True,
    )

    legacy_path = PythonOperator(task_id="legacy_path", python_callable=lambda: None)
    new_path_a = PythonOperator(task_id="new_path_a", python_callable=lambda: None)
    new_path_b = PythonOperator(task_id="new_path_b", python_callable=lambda: None)

    join = PythonOperator(
        task_id="join",
        python_callable=lambda: None,
        trigger_rule="none_failed_min_one_success",   # tolerates the skipped branch(es)
    )

    branch >> [legacy_path, new_path_a, new_path_b] >> join

34Voluntary skip#

What it does: Raising PiFlowSkip("reason") inside a Python callable marks that task instance skipped instead of failed — the idiomatic way to say "there was nothing to do this run."

How it works:

• No retry is consumed, and on_failure_callback never fires for a voluntary skip — it goes through the skip path, firing on_skipped_callback instead (feature 21).
• Downstream tasks see this exactly like any other skipped state for trigger-rule purposes — a default all_success downstream task will be blocked/cascade-skipped just as if a branch had explicitly skipped it; use none_failed-family rules (feature 14) on anything that should still proceed.
• PiFlowSkip only works from inside the callable's own exception flow — raising it from a separately-spawned thread, or after the callable has already returned normally, has no effect.
• The reason string is written to the task's log but isn't stored in a separate queryable field — push it via ti.xcom_push or a task note first if you need it available for later reporting.

Example:

from dag_parser.dynamic.dag_context import PiFlowSkip

def maybe_process(**context):
    if context["ds"] in ("2026-01-01", "2026-12-25"):
        raise PiFlowSkip("holiday — nothing to process today")
    print("processing normally...")

conditional_task = PythonOperator(
    task_id="conditional_process",
    python_callable=maybe_process,
    provide_context=True,
)