Branching and Trigger Rules

Branching is the first moment an Airflow DAG stops being a straight line. Until now every task in the bookshop pipeline has run on every run — extract, transform, load, in order, no questions asked. But real pipelines have forks in them. Did new data actually arrive today, or is the source file the same one we processed yesterday? Is this a full reload or an incremental one? Did the freshness check pass, or do we need to page someone instead of loading garbage?

Each of those is a question the DAG has to answer while it runs, and then act on. Airflow’s model for this is deceptively simple: a task can look at the situation and decide which downstream path should run. Every path it doesn’t choose gets marked skipped. Skipped is not failed — the run is still healthy — but it is a real state that flows downstream, and it is the thing that trips up almost everyone the first time they put a join after a branch. This post is about making that fork work: how to write the branch, what skip actually does to the graph, and the trigger rules that let a downstream task survive a branch instead of being collateral damage.

The branch task: @task.branch

The Task SDK gives you a decorator that turns a function into a branch: @task.branch. It looks like an ordinary @task, but its return value is special. Instead of returning data, it returns the task_id (or a list of task ids) of whichever downstream task should run next. Everything else downstream of the branch is skipped.

from airflow.sdk import dag, task

@dag(schedule="@daily", catchup=False, tags=["bookshop"])
def bookshop_conditional():
    @task
    def count_new_orders() -> int:
        # returns how many order rows landed since the last run
        return 128

    @task.branch
    def choose_path(n: int) -> str:
        if n > 0:
            return "load_orders"
        return "skip_load"

    @task
    def load_orders():
        print("loading today's orders")

    @task
    def skip_load():
        print("nothing to load today")

    n = count_new_orders()
    branch = choose_path(n)
    branch >> [load_orders(), skip_load()]

bookshop_conditional()

Read the graph the branch builds. choose_path sits upstream of both load_orders and skip_load. At run time it returns exactly one of those two ids as a string. The task whose id matches runs; the other one — and, crucially, everything downstream of it — is skipped. The string you return must match a real, directly-downstream task_id, or Airflow raises an error at run time. That last constraint catches people: a branch can only steer to tasks that are actually wired below it, not to some arbitrary task elsewhere in the DAG.

You can return a list when more than one path should run:

@task.branch
def choose_path(n: int) -> list[str]:
    if n > 1000:
        return ["load_orders", "alert_high_volume"]
    if n > 0:
        return ["load_orders"]
    return ["skip_load"]

Now a big day runs both the load and an alert; an ordinary day runs just the load; an empty day runs the skip branch. The tasks you didn’t name are skipped either way.

There’s a classic form too, in case you meet it in an older codebase or need an operator rather than a decorated function: BranchPythonOperator, which in Airflow 3.x lives at airflow.providers.standard.operators.python.

from airflow.providers.standard.operators.python import BranchPythonOperator

def _choose(**context):
    n = context["ti"].xcom_pull(task_ids="count_new_orders")
    return "load_orders" if n > 0 else "skip_load"

choose = BranchPythonOperator(task_id="choose_path", python_callable=_choose)

Same behavior, more ceremony: you pull the upstream value out of XCom yourself and return a task id. For new code, @task.branch is the one to reach for — it reads like the TaskFlow you already know. The operator is worth recognizing but not worth choosing.

What “skipped” actually does

Here is the mechanism that everything else in this post depends on. When a branch picks a path, the tasks on the roads-not-taken don’t just sit idle — Airflow actively sets them to skipped, and that skip cascades downstream. If skip_load is skipped and it has three tasks below it, those three inherit the skip too. Skip is contagious; it flows down the graph exactly the way success does.

This is by design, and it’s the right design. If the branch decided the load path shouldn’t run, you certainly don’t want the reporting task that sits below the load path to run either. The skip propagating down the branch is Airflow keeping the whole subtree consistent with the decision made at the fork.

The catch is what happens when the two branches come back together. A task that sits below both the load path and the skip path — a join — has one skipped parent no matter which way the branch went. And the default rule for whether a task may start is all_success, which reads its parents strictly: every upstream must have succeeded, and a skipped parent is not a success. So the join gets skipped as well, and skip keeps cascading, and you discover that adding an innocent “send the summary” task at the bottom of your DAG quietly turned itself off. That’s the join problem, and the fix is trigger rules — but first, the whole menu of them.

@task.short_circuit: the branch’s simpler cousin

Before the full rules, one more branching primitive, because it answers a narrower question more cleanly. Sometimes you don’t need to choose between paths — you just need to decide whether to keep going at all. “If the freshness check fails, stop; otherwise run the rest of the DAG.” That’s @task.short_circuit.

@task.short_circuit
def data_is_fresh(path: str) -> bool:
    # returns True to continue, False to short-circuit
    return file_modified_today(path)

fresh = data_is_fresh(source_path)
fresh >> transform() >> load()

A short-circuit task returns a boolean. Return something truthy and the DAG carries on as normal. Return something falsy and everything downstream is skipped — you don’t name task ids, you don’t build two branches, you just stop the flow here. It’s the natural tool for guard clauses: a validation gate, a “did the upstream job even land a file” check, a feature flag.

By default the skip from a short-circuit propagates all the way down, respecting each downstream task’s trigger rule as it goes. If you have a cleanup task at the very bottom set to run all_done, you might want it to still fire even when the short-circuit stopped the pipeline. The knob for that is ignore_downstream_trigger_rules, which defaults to True:

@task.short_circuit(ignore_downstream_trigger_rules=False)
def data_is_fresh(path: str) -> bool:
    return file_modified_today(path)

With the default (True), a falsy return skips all downstream tasks regardless of their own trigger rules — the short-circuit wins outright. Set it to False and the skip only reaches the tasks that would themselves be skipped under normal propagation, letting a downstream all_done cleanup task still run. Most of the time the default is what you want; the flag exists for the case where a teardown or notification must happen even on a short-circuit.

The full trigger-rule set

A trigger rule answers one question for each task: given the states of my direct upstream tasks, am I allowed to start? The default, all_success, is the strict one — everybody upstream must have succeeded. But there are ten more, and each exists for a real situation. Here’s the whole set, with the states each one cares about:

Trigger rule	Task runs when…	Reach for it when
all_success (default)	every upstream succeeded	the normal case — a task that needs all its inputs
all_failed	every upstream failed	a fallback that only makes sense if everything before it broke
all_done	every upstream finished, any state	cleanup and teardown that must run win or lose
all_skipped	every upstream was skipped	a “nothing happened” path after a branch
one_success	at least one upstream succeeded (fires early)	a join over redundant sources — first good one wins
one_failed	at least one upstream failed (fires early)	an alert/handler that trips the moment anything breaks
one_done	at least one upstream succeeded or failed	react as soon as any branch resolves, ignoring skips
none_failed	no upstream failed; skips are OK	a join after a branch where failure is still fatal
none_failed_min_one_success	no upstream failed and at least one succeeded	the branch-join fix — the one you’ll use most
none_skipped	no upstream was skipped; success/fail OK	a task that must not run if any path was pruned
always	always, ignoring upstream entirely	a task with no real dependency on what came before

Two subtleties make this table make sense. First, there’s a hidden state you haven’t met yet: upstream_failed. When a task’s parents can never satisfy its trigger rule — say an all_success join whose parent already failed — Airflow doesn’t leave it stuck; it marks it upstream_failed and moves on. So “failed upstream” in these rules covers both a task that actually failed and one that was marked upstream_failed because its parents failed. Second, notice how the rules split into two families: the all_*/none_* family waits for every parent to finish before deciding, while the one_* family can fire the instant a single parent reaches the state it’s watching for — useful when you want to react quickly and don’t care about the stragglers.

The two you’ll actually type most days are all_done (for cleanup that has to run regardless) and none_failed_min_one_success (for the join after a branch). The rest are there for the day you need them, and it’s worth knowing they exist so you recognize the shape of the problem when you hit it.

Fixing the join

Back to the broken join. Recall the setup: a branch chooses between a load path and a skip path, and a report task sits below both to summarize the run. With the default all_success, report sees one skipped parent every single run and skips itself. Here’s the whole thing, fixed:

from airflow.sdk import dag, task

@dag(schedule="@daily", catchup=False, tags=["bookshop"])
def bookshop_conditional():
    @task
    def count_new_orders() -> int:
        return 128

    @task.branch
    def choose_path(n: int) -> str:
        return "load_orders" if n > 0 else "skip_load"

    @task
    def load_orders():
        print("loaded today's orders")

    @task
    def skip_load():
        print("no new orders today")

    @task(trigger_rule="none_failed_min_one_success")
    def report():
        print("run summary written")

    n = count_new_orders()
    branch = choose_path(n)
    loaded = load_orders()
    skipped = skip_load()
    branch >> [loaded, skipped]
    [loaded, skipped] >> report()

bookshop_conditional()

The only change that matters is trigger_rule="none_failed_min_one_success" on report. Now walk the two outcomes:

• New orders arrived. choose_path returns "load_orders". load_orders runs and succeeds; skip_load is skipped. At the join, report sees one success and one skip, no failures — none_failed_min_one_success is satisfied, so report runs.
• No new orders. choose_path returns "skip_load". skip_load runs and succeeds; load_orders is skipped. report again sees one success, one skip, no failures — it runs.

Either way exactly one branch succeeds, the other is skipped, no task fails, and report fires. That’s precisely what none_failed_min_one_success encodes: don’t run if anything upstream broke, but a skip is fine as long as at least one real thing succeeded. It is the canonical branch-join rule, and once you’ve been burned by the silent-skip once you’ll reach for it on reflex.

Why none_failed_min_one_success and not plain none_failed? none_failed would also run report when both parents were skipped — which can happen if the branch itself was skipped by something further up. The min_one_success clause insists that at least one upstream genuinely did work, so report only summarizes runs where something actually happened. If you’d rather the report run even in the all-skipped case, none_failed is your rule instead. That’s the granularity these eleven rules buy you.

Trigger rules in the wild: alerts and cleanup

The join fix is the rule you’ll reach for most, but two others earn their keep in almost every production DAG, and they’re worth seeing worked out because they solve problems the default rule actively gets wrong.

The first is failure handling. Suppose the bookshop load runs three tasks in parallel — orders, inventory, reviews — and if any of them breaks, you want to fire a task that posts to Slack and quarantines the partial data. You can’t hang that task off the pipeline with the default rule: all_success would demand all three succeed, which is the exact opposite of when you want the alert. one_failed is the rule, and it fires the moment the first of the three fails, without waiting for the others:

@task
def load_orders(): ...

@task
def load_inventory(): ...

@task
def load_reviews(): ...

@task(trigger_rule="one_failed")
def on_failure_alert():
    print("a load failed — posting to Slack and quarantining")

loads = [load_orders(), load_inventory(), load_reviews()]
loads >> on_failure_alert()

On a clean run all three succeed, one_failed is never satisfied, and on_failure_alert is skipped — exactly right, no alert on success. Break one and the alert trips immediately. This is the pattern behind most “notify on failure” wiring, and it composes with everything else: on_failure_alert is just a task, so it can have its own downstream teardown.

The second is cleanup that must always run. A task that drops a temp table, releases a lock, or tears down a scratch directory has to fire whether the pipeline succeeded, failed, or was short-circuited. That’s all_done — it waits for every upstream to finish, in any state, then runs:

@task(trigger_rule="all_done")
def drop_scratch_tables():
    print("cleaning up temp tables regardless of outcome")

loads >> drop_scratch_tables()

Because all_done ignores upstream state and only cares that everyone is finished, the cleanup runs on the green path, on the failed path, and on the skipped path alike. Pair it with the ignore_downstream_trigger_rules=False short-circuit setting from earlier and you can guarantee a teardown fires even when a guard clause stopped the whole run. One caution: because all_done runs regardless of failure, don’t put real pipeline logic behind it — a downstream all_success task after an all_done cleanup will happily proceed even though the pipeline it was cleaning up after actually failed. Keep all_done for teardown, not for work.

Two rules, two jobs the default could never do: one_failed to catch a break the instant it happens, all_done to guarantee cleanup no matter what. Between those, none_failed_min_one_success for the branch join, and all_success for everything ordinary, you’ve covered the vast majority of real DAGs.

Labeling the edges

A branch makes the graph harder to read, because now an edge means “the path taken when the branch chose this.” Airflow lets you annotate edges with Label, from the Task SDK, so the graph in the UI explains itself:

from airflow.sdk import Label

branch >> Label("new data") >> loaded
branch >> Label("nothing new") >> skipped

Label is pure documentation — it changes nothing about execution, it just prints text on the edge in the graph view. But on a DAG with real branching it’s the difference between a maze and a diagram. When someone opens the UI at 3 a.m. to see why the load didn’t run, “nothing new” written on the edge into the skip path answers the question before they’ve clicked anything. Label the edges out of every branch; your future on-call self will thank you.

The bookshop pipeline, with a real fork

Let’s put it all together into a pipeline that decides, each day, whether there’s work to do — and stays green whichever way it goes.

from airflow.sdk import dag, task, Label

@dag(schedule="@daily", catchup=False, tags=["bookshop"])
def bookshop_daily_conditional():
    @task
    def check_source() -> int:
        # count order rows that landed since the last successful run
        return new_row_count()

    @task.branch
    def route(n: int) -> str:
        # steer to the load path only if there's something to load
        return "load_path.transform" if n > 0 else "no_op"

    @task
    def no_op():
        print("no new orders — nothing to do today")

    from airflow.sdk import task_group

    @task_group(group_id="load_path")
    def load_path():
        @task
        def transform() -> str:
            return "/data/orders/clean.parquet"

        @task
        def load(path: str):
            print(f"loaded {path} into DuckDB")

        load(transform())

    @task(trigger_rule="none_failed_min_one_success")
    def write_run_report():
        print("run summary written")

    n = check_source()
    branch = route(n)
    work = load_path()
    nothing = no_op()

    branch >> Label("new data") >> work
    branch >> Label("no work") >> nothing
    [work, nothing] >> write_run_report()

bookshop_daily_conditional()

Three things worth calling out. The branch steers to load_path.transform — the first task inside the TaskGroup — because that’s the real entry point of the load path; a branch has to name an actual downstream task, and the group id alone isn’t one. When there’s nothing to do, the whole load_path group is skipped in one go, TaskGroups and all, because the skip cascades through the group just like it would through loose tasks. And write_run_report closes the fork with none_failed_min_one_success, so the daily summary is written on every run — busy day or quiet one — without you ever seeing a spurious skip. Open this DAG in the UI and the labeled edges tell the whole story at a glance: new data goes left into the load, no work goes right into the no-op, and both roads lead to the report.

A few sharp edges

Once you start using branches in anger, a handful of gotchas show up often enough to name.

A branch should only point at its direct children — and the way it fails when you get this wrong is nastier than an error. The task id you return is meant to name a task wired immediately downstream of the branch. Airflow validates the returned id against every task in the DAG, not against the branch’s children, and those are two very different checks:

• Return an id that doesn’t exist in the DAG at all and you get the loud failure you’d hope for: AirflowException: 'branch_task_ids' must contain only valid task_ids. Invalid tasks found: {'does_not_exist'}. Fine — a typo, caught.
• Return an id that exists in the DAG but isn’t a direct child — a task two hops down, or one on a different limb — and validation passes. The branch task goes green. Then Airflow skips every one of the branch’s actual children, because none of them matched. No error, no warning: a successful branch that silently skips the entire fork beneath it.

That second case is the one that costs you a morning, and it’s a natural mistake — “I want the load step to run, so I’ll return load” is a perfectly reasonable thing to think when load sits below stage. Say it out loud instead: a branch steers the fork immediately below it, and nothing further. If you need to reach deeper, branch to the entry task of that subtree and let the rest follow by ordinary dependency. And if a branch DAG ever runs green while doing nothing at all, this is the first thing to check.

Skip propagation stops at a tolerant trigger rule. Skip cascades down through every all_success task in its path, but the moment it reaches a task whose rule accepts a skip — none_failed_min_one_success, all_done, one_success — the cascade halts there and that task evaluates normally. This is exactly why the join fix works: the tolerant rule is a firebreak that stops the skip from burning through the rest of the DAG. Place that firebreak deliberately, at the join, not three tasks too late.

Don’t nest branches without a plan. A branch downstream of another branch is legal, but the skip states multiply and the graph gets genuinely hard to reason about. More often what you actually wanted was a single branch returning a list of task ids, or a short-circuit guarding a whole subtree. Reach for those before you stack forks on forks.

Skip is not the same as retry. A skipped task never ran and won’t retry — retries only apply to a task that started and failed. If you find yourself wishing a skipped task would “try again,” the decision belongs in the branch logic, not in a retry policy. The branch is where the choice lives; downstream tasks only ever inherit it.

Final thoughts

Branching is only hard for as long as skip surprises you. Once you internalize that a not-taken path is set to skipped, that skip cascades downstream the same way success does, and that the default all_success treats a skipped parent as a failure to launch, every strange “why didn’t my join run” mystery dissolves into the same fix: give the join a trigger rule that tolerates skips. none_failed_min_one_success is that rule nine times out of ten, all_done is the one for cleanup that must always run, and the other nine are there for the specific day you need them. Treat the trigger rule as part of a task’s contract — it’s as load-bearing as the dependencies themselves — and label your branch edges so the graph reads like the decision it encodes. Do that, and a forking DAG is no harder to reason about than a straight one; it just has more to say.

Next: Dynamic Task Mapping and TaskGroups