Airflow Conditional Execution: Yesterday's Data, Today's Run

Your DAG fired at 9:02 AM. The run row is green, every task shows success. And the dashboard is showing yesterday's numbers anyway. Sound familiar? That's the gap between "the DAG ran" and "the data is fresh" — and it's the gap that gets teams in trouble.

Schedules drift. The source feed lands late, a worker lags, someone hits the trigger button outside the window. When the run finally happens, nothing stops it from reprocessing a data interval that went stale hours ago. The scheduler doesn't care about freshness. It only cares about execution.

That's what conditional execution is for: skip semantics, ShortCircuitOperator, LatestOnlyOperator, and guards that read data_interval_end before touching any data. Each one answers a different version of "should this run happen at all?". This article is about choosing the right one — and knowing when you shouldn't be conditionally executing in the first place.

1. Why "The DAG Ran" Doesn't Mean the Data Is Fresh

Airflow is an execution engine, not a freshness engine. A DAG run fires on a schedule, executes tasks, and records states. Nothing in that loop knows whether the data inside the run's interval is what the business wants on screen right now.

The mismatch appears the moment reality breaks the schedule: a late file, a slow worker, a holiday, a manual trigger. The run executes anyway, against the interval it was created for. If that interval isn't today's, the run is quietly serving old data while every status light stays green.

This is why the most senior habit in this article is skepticism: never read run success as data success. Read the interval.

from datetime import datetime, timezone

from airflow.decorators import dag, task
from airflow.models import DAGRun


def assert_fresh_interval(dag_run: DAGRun) -> None:
    end = dag_run.data_interval_end
    today = datetime.now(timezone.utc).date()
    if end.date() != today:
        raise ValueError(
            f"Stale interval: run ended {end.isoformat()}, expected today."
        )


@task
@dag(schedule="@daily")
def stale_check():
    pass

2. ShortCircuit vs Branch: Two Ways to Skip

BranchPythonOperator returns a task_id and sends the run down one path. ShortCircuitOperator returns a boolean: True keeps the whole downstream chain, False skips everything after it. Branch chooses between paths; ShortCircuit decides whether the journey happens at all.

That distinction drives production design. Use Branch when two or more real outcomes exist — say, "file is full" vs "file is empty". Use ShortCircuit when there is exactly one outcome that should do work, and every other outcome should do nothing.

The classic trap: teams use Branch for a yes/no question, then fight the skipped branch through every join. A boolean question deserves a ShortCircuit, not a fork in the road.

One behavior to know before you rely on a short-circuit: its skip overrides downstream trigger rules by default. A join with none_failed sitting after a short-circuited chain is still skipped — even when another branch of the graph succeeded. Set ignore_downstream_trigger_rules=False to let downstream trigger rules evaluate again; only the direct children stay skipped.

from datetime import datetime, timedelta, timezone

from airflow.decorators import dag
from airflow.models import DAGRun
from airflow.operators.python import ShortCircuitOperator


def is_today_interval(dag_run: DAGRun) -> bool:
    end = dag_run.data_interval_end
    return end.date() == datetime.now(timezone.utc).date()


@dag(
    schedule="@daily",
    start_date=datetime(2026, 7, 1),
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
)
def daily_revenue_etl():
    guard = ShortCircuitOperator(
        task_id="guard_fresh_interval",
        python_callable=is_today_interval,
    )

    @task
    def load_revenue() -> str:
        return "revenue loaded"

    guard >> load_revenue()


daily_revenue_etl()

3. LatestOnlyOperator and Late Schedules

LatestOnlyOperator sits at the head of a DAG and lets exactly one run through: the latest. When the scheduler fires a run for an old interval — after an outage, a catchup, or a manual trigger — LatestOnly skips everything downstream for that old run.

It's the bluntest freshness tool, and it's perfect for exactly one scenario: a DAG that must only ever run for the current interval, period. Reconciliations and backfills need old intervals processed, and LatestOnly would skip them all.

Know the weapon before you wield it. LatestOnly is a one-liner that protects a whole chain, but it quietly disables history.

from datetime import datetime, timedelta

from airflow.decorators import dag, task
from airflow.operators.latest_only import LatestOnlyOperator


@dag(
    schedule="@daily",
    start_date=datetime(2026, 7, 1),
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
)
def latest_dashboard():
    gate = LatestOnlyOperator(task_id="latest_only_gate")

    @task
    def refresh_dashboard() -> str:
        return "dashboard refreshed"

    gate >> refresh_dashboard()


latest_dashboard()

4. Guarding on data_interval_end

Data intervals are the real contract of a run. Every run carries data_interval_start and data_interval_end, and the scheduler picks these from the schedule, not from the wall clock. A guard that compares data_interval_end to today's date is the most precise freshness check you can write.

Where do you put it? Either in a ShortCircuitOperator callable, as in section 2, or directly inside the load task as a hard assertion. The assertion version is stronger: it fails loudly instead of skipping silently, which is what you want for compliance-grade loads.

One warning: the first run for a schedule fires after start_date plus one interval, and timezone boundaries can shift the comparison by a day. Always compare in UTC against the run's own interval, never against the local clock of the worker.

from datetime import datetime, timedelta, timezone

from airflow.decorators import dag, task
from airflow.models import DAGRun


@dag(
    schedule="@daily",
    start_date=datetime(2026, 7, 1),
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
)
def guarded_load():
    @task
    def load_revenue(dag_run: DAGRun | None = None) -> str:
        end = dag_run.data_interval_end
        if end.date() != datetime.now(timezone.utc).date():
            raise ValueError(f"Refusing stale interval ending {end.isoformat()}")
        return "rows upserted for " + end.date().isoformat()

    load_revenue()


guarded_load()

5. Building Conditional Task Chains

Conditional execution changes the state graph: skipped tasks appear, and joins must survive them. The default trigger rule all_success fails a join task the moment any upstream is skipped. That's the exact failure from the branching article, and it recurs in every conditional DAG until you set rules deliberately.

The production pattern is simple: guards at the head, joins with none_failed or all_done, and leaf tasks that read skip states before writing. A chain that treats skipped as a first-class state is a chain you can debug in the Grid view.

When you find yourself adding the third condition, stop. Three booleans in one DAG usually mean three DAGs, each with one job and one guard.

6. Idempotent Guards Instead of Conditional Loads

Here's the opinionated part: the best conditional execution is the one you don't need. If a load is fully idempotent — upsert on a natural key, delete-and-reload a partition — then replaying an interval is harmless, and the freshness guard becomes a nicety instead of a lifeline.

Teams that condition their loads instead of making them idempotent end up with branches for every edge case: partial files, re-sends, holiday skips. Each branch is a place where the wrong state can slip through. Each branch is also dead weight when the real fix is an upsert.

So the senior ordering is: make the load idempotent first, add the interval guard second, and keep the DAG linear. Conditional execution is a shield, not a strategy.