Airflow Conditional Execution: Stop Stale Data Fast

Your daily DAG ran on time and loaded yesterday's file again. The schedule was right, but the data was stale and nobody checked.

You'll add guards that stop stale runs cold. ShortCircuit for freshness, LatestOnly for catchup, interval checks for precision.

We'll show which guard fits each case and how to test late schedules without waiting for midnight. Freshness becomes enforced, not hoped for.

Run means fresh. Or skip.

Why DAG Ran Does Not Mean Data Is Fresh

Schedules fire on time even when inputs are late. Without a guard, the load reads whatever partition exists and labels it current.

Freshness must be checked per run against the interval, not assumed from the clock. That check belongs in the DAG as code.

ShortCircuit vs Branch

ShortCircuitOperator runs downstream only when its callable returns True, otherwise it skips everything below. Branch picks one of several paths.

Use ShortCircuit for go or no-go freshness gates. Use Branch when you have two valid paths like weekday versus weekend logic.

Two details bite everyone once. ShortCircuit pushes its return value to XCom on True, so downstream can read the gate's verdict, and any falsy value short-circuits: note [False] is truthy in Python, so return a bare False, never a one-element list. Prefer @task.short_circuit for new code; same semantics, decorator style. For partial short-circuits, set ignore_downstream_trigger_rules False: direct children skip while deeper tasks honor their own rules, so an all_done notifier at the tail still fires. Airflow 3.x adds @task.run_if and @task.skip_if for runtime conditions without restructuring the DAG; use those for simple predicates and keep ShortCircuit for gates with rich Python logic.

from airflow.decorators import dag, task
from airflow.operators.python import ShortCircuitOperator
from datetime import datetime

def _is_fresh(**ctx):
    key = ctx["data_interval_end"].strftime("inventory/%Y-%m-%d.csv")
    import boto3
    s3 = boto3.client("s3")
    try:
        s3.head_object(Bucket="retail-landing", Key=key)
        return True
    except Exception:
        return False

@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False, tags=["retail"])
def inventory_daily():
    guard = ShortCircuitOperator(task_id="guard_freshness", python_callable=_is_fresh)

    @task
    def load_inventory():
        return {"status": "loaded"}

    guard >> load_inventory()

inventory_daily()
# return False (bare) to skip; [False] is truthy and WON'T skip
# @task.short_circuit is the modern spelling
# ignore_downstream_trigger_rules=False lets a tail all_done notifier still fire

LatestOnly and Late Schedules

LatestOnlyOperator runs its downstream only on the latest scheduled interval. During catchup or backfill, older intervals skip the gated branch.

Place it before expensive or freshness-sensitive work like full refreshes. Cheap idempotent history loads stay outside the gate.

LatestOnly never fires downstream on externally triggered runs either, since a manual trigger isn't the latest interval; plan a manual override path for backfills that genuinely need the expensive branch. It also pairs badly with catchup True pipelines unless the gated work is truly latest-only like dashboard refreshes.

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

@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=True, tags=["retail"])
def inventory_latest():
    latest = LatestOnlyOperator(task_id="latest_only")

    @task
    def refresh_dashboard():
        return {"refreshed": True}

    @task
    def archive_history():
        return {"archived": True}

    latest >> refresh_dashboard()
    archive_history()

inventory_latest()

Guarding on Data Interval End

data_interval_end is the scheduling anchor for a run. Derive partition keys and watermark queries from it, never from datetime.now.

Backfills and late runs then read the correct slice automatically. Wall-clock logic breaks the moment a run is not on time.

Conditional Task Chains

Chain guards so freshness, quality, and cost checks compose. A ShortCircuit for input presence feeds a quality check that feeds the load.

Keep each guard tiny and observable. Log why a run skipped so on-call knows whether to page the vendor or ignore it.

Terminate every chain with an all_done notifier that logs the skip reason and the interval; silent skips rot into mystery pages three months later. Keep guards ordered cheapest-first so a cheap partition check runs before the pricey warehouse probe.

from airflow.decorators import dag, task
from datetime import datetime

@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False, tags=["retail"])
def inventory_chain():
    @task.short_circuit
    def input_present(data_interval_end=None):
        key = data_interval_end.strftime("%Y-%m-%d")
        return {"key": key} if key else False

    @task
    def load_partition(key: str = ""):
        return {"loaded": key}

    @task(trigger_rule="all_done")
    def notify_skip():
        return {"notified": True}

    gate = input_present()
    gate >> load_partition()
    gate >> notify_skip()

inventory_chain()

Idempotent Guards Instead of Conditional Loads

The strongest guard is a load that can safely rerun. Upsert by interval key or partition overwrite makes reruns harmless.

Conditional logic then handles notification and cost, while idempotency handles safety. You get both cheap reruns and correct skips.