How to run a task only if a condition is true in Airflow (branching)

Airflow has three tools for conditional tasks. @task.run_if skips a single task, @task.short_circuit stops everything after it, and @task.branch picks which of several paths runs. Branching needs one extra setting on the task where the paths meet again, or that task is skipped as well.

Copy and paste

Run one task only when a condition is true. Here, only on the first day of the month:

from airflow.sdk import task

@task.run_if(lambda context: context["ds"].endswith("-01"))
@task
def monthly_report():
    ...

@task.skip_if is the opposite: it skips the task when the condition is true.

Stop the rest of the DAG when a condition is false:

@task.short_circuit
def has_new_files():
    files = list_new_files()   # your own check
    return len(files) > 0

has_new_files() >> process_files() >> publish()

Choose between two paths:

from airflow.sdk import task

@task.branch
def choose_load(ds=None):
    if ds.endswith("-01"):
        return "full_load"           # the task_id to run
    return "incremental_load"

@task
def full_load():
    ...

@task
def incremental_load():
    ...

@task(trigger_rule="none_failed_min_one_success")
def publish():
    ...

choose_load() >> [full_load(), incremental_load()] >> publish()

What happens in each run

The branching DAG above on two dates, with a second publish task that keeps the default trigger rule for comparison:

The task after a branch needs a trigger rule

By default a task runs only when all its upstream tasks succeeded (all_success). After a branch one path is always skipped, so a join task with the default rule is skipped on every run. none_failed_min_one_success runs it when no upstream task failed and at least one succeeded. See how to run a task even if the previous one failed for the other rules.

Branch details

• Return one task_id, a list of task IDs to run several paths, or None to skip every task directly after the branch.
• The IDs must belong to tasks directly downstream of the branch task. Airflow does not raise an error for any other ID: it skips the direct downstream tasks, and the task you named is skipped with them. Inside a task group, include the group prefix: "loads.full_load".
• The classic operator is BranchPythonOperator, imported from airflow.providers.standard.operators.python.

short_circuit or run_if

When @task.short_circuit returns a false value, it skips every task downstream, whatever their trigger rules. Pass ignore_downstream_trigger_rules=False to skip only the tasks directly after it and let the trigger rules decide for the rest.

@task.run_if and @task.skip_if act on one task. The condition receives the context, so it can read ds, params or dag_run.conf. Tasks downstream of a skipped task follow their trigger rules: with the default rule, they are skipped too.

Tested with Airflow 3.3.2.