Airflow DAGs skill for Claude Code

A DAG is done when the file parses in well under a second with no I/O, every task reads its window from the run (never the wall clock) and can be rerun without duplicates, waiting happens in the triggerer instead of a worker slot, every task and run has a timeout, and an integrity test plus one dags test run pass in CI.

First, identify the setup

airflow version                                  # 3.x? which minor?
pip list 2>/dev/null | grep -E "apache-airflow(-providers-[a-z-]+)? "
airflow config get-value core executor
airflow config get-value scheduler create_cron_data_intervals
airflow dags list-import-errors

Managed services (MWAA, Cloud Composer, Astronomer) pin their own Airflow and provider versions; check the environment’s version before using a 3.x-only feature.

Airflow 3 facts that change how you write DAGs

• Author with airflow.sdk: dag, task, DAG, Asset, AssetAlias, Variable, Param, TaskGroup, get_current_context. Core operators and sensors moved to airflow.providers.standard (for example airflow.providers.standard.operators.python). Old paths warn in 3.1 and will be removed.
• Task code runs through the Task SDK and cannot open the metadata database (no settings.Session, no ORM models in tasks). Use the REST API client instead.
• Removed: schedule_interval (use schedule), SLAs (sla=, sla_miss_callback; use Deadline Alerts, 3.1+), SubDAGs (use TaskGroups or assets), and context keys execution_date, next_ds, prev_ds, yesterday_ds, tomorrow_ds and their variants. Use logical_date and data_interval_start / data_interval_end.
• catchup defaults to False.
• Bare cron strings and presets now resolve to CronTriggerTimetable because [scheduler] create_cron_data_intervals defaults to False. For those runs data_interval_start == data_interval_end. Code that computed “yesterday” from the interval silently processes a zero-length window. See Scheduling.
• xcom_pull() without task_ids reads only the current task.
• Datasets are Assets; @asset defines a one-task DAG producing an asset.
• Airflow 3.2 added asset partitioning (CronPartitionTimetable, PartitionedAssetTimetable, dag_run.partition_key); check your minor version before relying on it.

Migrating a 2.x DAG folder: ruff check dags/ --select AIR301,AIR302 (breaking) and AIR311,AIR312 (recommended), ruff 0.13.1 or newer; add --fix after reviewing.

Workflow

- [ ] 1. Define the run unit: what one DAG run processes (a day, an hour, a partition)
- [ ] 2. Pick the timetable or asset trigger that matches that unit
- [ ] 3. Write tasks as idempotent functions of the run's window
- [ ] 4. Set retries, execution_timeout, dagrun_timeout, max_active_runs, pools
- [ ] 5. Make waits deferrable; give every sensor a timeout
- [ ] 6. Keep the file parse-cheap; run scripts/check_dag_files.py
- [ ] 7. Integrity test + `airflow dags test` for one logical date
- [ ] 8. Backfill a small range, verify, then the full range (user confirms range)

Scheduling

Need	Schedule
Process the previous day/hour as a window	CronDataIntervalTimetable("0 2 * * *", timezone="UTC") (interval = previous cron period)
Fire at a time, window not needed (reports, pings)	Cron string or CronTriggerTimetable(...)
Fire at a time but process a fixed lookback	CronTriggerTimetable("0 2 * * *", timezone="UTC", interval=timedelta(days=1))
Fixed spacing regardless of clock	DeltaTriggerTimetable(timedelta(hours=6))
Run when upstream data lands	schedule=[orders_asset], or (a & b), `(a
Both asset updates and a clock	AssetOrTimeSchedule(...)
Only manual or API triggered	schedule=None

• Pass a timetable object rather than relying on the create_cron_data_intervals setting, so a config change cannot alter logical_date semantics.
• Switching an existing DAG from a trigger to a data-interval timetable skips one run. For big schedule changes, create a new dag_id.
• Asset events are emitted only when the producing task succeeds. Cross-DAG timing by guessing (“run at 03:00, upstream usually finishes by 02:45”) is a bug; use assets.
• Schedules, start_date, and intervals in UTC. Use a fixed pendulum.datetime(2026, 1, 1, tz="UTC"), never datetime.now() or days_ago().

A reference DAG

from datetime import timedelta
import pendulum
from airflow.sdk import Asset, CronDataIntervalTimetable, dag, task

raw_orders = Asset("s3://lake/raw/orders")

@dag(
    schedule=CronDataIntervalTimetable("0 2 * * *", timezone="UTC"),
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    max_active_runs=1,
    dagrun_timeout=timedelta(hours=3),
    default_args={
        "owner": "data-eng",
        "retries": 3,
        "retry_delay": timedelta(minutes=5),
        "retry_exponential_backoff": True,
        "execution_timeout": timedelta(minutes=45),
    },
    tags=["orders", "tier-1"],
)
def orders_daily():
    @task(outlets=[raw_orders])
    def extract(data_interval_start=None, data_interval_end=None) -> str:
        from my_pkg.orders import export_orders          # heavy imports inside the task
        day = data_interval_start.strftime("%Y-%m-%d")
        path = f"s3://lake/raw/orders/dt={day}/part.parquet"
        export_orders(start=data_interval_start, end=data_interval_end, dest=path)  # overwrite
        return path                                       # small reference, not data

    @task(pool="warehouse", retries=1)
    def load(path: str, data_interval_start=None) -> int:
        from my_pkg.warehouse import replace_partition
        return replace_partition("raw.orders", path, dt=data_interval_start.date())

    @task(retries=0)
    def validate(rows: int) -> None:
        if rows == 0:
            raise ValueError("0 rows loaded")             # fail fast; retrying will not help

    validate(load(extract()))

orders_daily()

Idempotency patterns (overwrite partition, MERGE on key, delete+insert in one transaction) are covered tool-agnostically in data-pipelines; in Airflow the key is that every task derives its window from data_interval_* or logical_date.

Parse-time hygiene

The DAG processor re-executes every file’s top-level code continuously.

• No network, database, Variable.get(), or cloud SDK calls at module level. Read Variables inside tasks or via templates: "{{ var.value.bucket }}".
• Import pandas, Spark clients, and SDKs inside the task function.
• Measure: time python dags/orders_daily.py minus time python -c ''. Aim for well under 1 s per file; [core] dagbag_import_timeout (default 30 s) kills slow files and the DAG disappears from the UI.
• Dynamic DAG generation from a config file in the repo is fine; generating from an API call at parse time is not.
• Run the bundled checker on every change:

python3 scripts/check_dag_files.py dags/            # errors exit 1
python3 scripts/check_dag_files.py dags/ --strict   # warnings also fail (CI)

It flags removed Airflow 2 parameters and context keys, dynamic start_date, legacy imports, top-level I/O, wall-clock calls inside tasks, sensors without timeout, and xcom_pull() without task_ids. It parses with ast and never imports the DAG.

Data passing and fan-out

• XCom (including TaskFlow return values) is for small metadata: paths, IDs, row counts. Large values bloat the metadata DB. For bigger payloads write to object storage and pass the URI, or configure the object storage XCom backend: [core] xcom_backend = airflow.providers.common.io.xcom.backend.XComObjectStorageBackend with [common.io] xcom_objectstorage_path and xcom_objectstorage_threshold (bytes; smaller values stay in the DB).
• Dynamic task mapping: process.expand(path=list_files()), constants via .partial(...), several args via .expand_kwargs(...). [core] max_map_length (default 1024) caps the fan-out; an empty input marks the mapped task skipped. Limit parallelism with max_active_tis_per_dag on the mapped task, and set map_index_template so UI rows are readable.
• Group related tasks with @task_group; do not nest DAGs.

Retries, timeouts, and deadlines

Setting	Rule
retries + retry_exponential_backoff	2-3 for tasks hitting networks or warehouses; retries=0 for validation and non-idempotent side effects
execution_timeout	On every task (via default_args); a hung task otherwise holds a slot forever
dagrun_timeout	Upper bound for the whole run, so a stuck run cannot block max_active_runs=1
Sensor timeout	Always explicit; the default is 7 days
on_failure_callback	Alert after retries are exhausted, with DAG, task, logical date, and log URL
Deadline Alerts (3.1+)	Replace SLAs: alert when a run is not done N minutes after it was queued or scheduled

from datetime import timedelta
from airflow.sdk import DAG, AsyncCallback, DeadlineAlert, DeadlineReference
from airflow.providers.slack.notifications.slack_webhook import SlackWebhookNotifier

with DAG(
    dag_id="orders_daily",
    deadline=DeadlineAlert(
        reference=DeadlineReference.DAGRUN_QUEUED_AT,      # or DAGRUN_LOGICAL_DATE, FIXED_DATETIME(...)
        interval=timedelta(minutes=90),
        callback=AsyncCallback(SlackWebhookNotifier,
                               kwargs={"text": "{{ dag_run.dag_id }} missed its deadline"}),
    ),
    ...
):
    ...

AsyncCallback runs in the triggerer (restart it when callback code changes); SyncCallback (3.2+) runs on an executor.

Waiting: sensors and deferrable operators

Wait type	Use
External job, file, or API state for minutes to hours	Deferrable operator or sensor (deferrable=True): frees the worker slot, waits in the triggerer
Short check (seconds)	Plain operator with retries
Sensor without a deferrable variant	mode="reschedule" with poke_interval of minutes, plus timeout
Upstream DAG’s data	Asset schedule, not an ExternalTaskSensor

• Deferrable requires at least one running airflow triggerer. Without it, deferred tasks sit in the deferred state forever.
• [operators] default_deferrable = True flips the default for providers that support it.
• Custom triggers: subclass BaseTrigger, implement serialize() and an async run() that yields TriggerEvent; no blocking calls and no side effects, because a trigger can run more than once.

Concurrency

• max_active_runs=1 when runs share mutable state or must apply in order.
• Pools (pool="warehouse", pool_slots=) protect shared systems across DAGs; size the pool to what the downstream system tolerates.
• max_active_tasks caps running task instances across all active runs of the DAG; priority_weight orders queued tasks when slots are scarce.

Backfills

airflow backfill create --dag-id orders_daily \
  --from-date 2026-09-01 --to-date 2026-09-07 \
  --reprocess-behavior failed --max-active-runs 2

• --reprocess-behavior: none (only missing runs), failed (missing and failed), completed (rerun everything in range). Use the dry-run option first to list dates.
• Ask the user to confirm the exact range and reprocess behavior before any backfill that rewrites existing partitions in production.
• Procedure: one date, verify output and row counts, a small range, check warehouse load and cost, then the full range with a low --max-active-runs.
• Confirm which code version the backfill runs (DAG versioning records the bundle version per run) and that old dates should be processed with that logic.

Testing

# tests/test_dag_integrity.py
import pytest
from airflow.dag_processing.dagbag import DagBag

@pytest.fixture(scope="session")
def dagbag():
    return DagBag(dag_folder="dags", include_examples=False)

def test_no_import_errors(dagbag):
    assert dagbag.import_errors == {}

def test_dag_conventions(dagbag):
    for dag_id, dag in dagbag.dags.items():
        assert dag.tags, f"{dag_id}: add tags"
        for t in dag.tasks:
            assert t.owner != "airflow", f"{dag_id}.{t.task_id}: set a real owner"
            assert t.execution_timeout is not None, f"{dag_id}.{t.task_id}: no execution_timeout"

airflow dags test orders_daily 2026-10-18           # full run in one process, real connections
airflow tasks test orders_daily load 2026-10-18     # one task, no state recorded
pytest tests/                                       # integrity + unit tests of task logic

Unit-test the business logic as plain functions (pass the window explicitly); the DAG should be thin wiring. dag.test() from __main__ gives the same full run inside an IDE debugger.

When not to use Airflow

Situation	Better fit
Asset/partition-first modeling, strong local testing, lineage as the core UI	Dagster
Python-native dynamic flows, little infrastructure, event-driven runs	Prefect
One script on a schedule with no dependencies	A scheduler (cron, cloud scheduler); see scheduled-jobs
Continuous event processing	A streaming system; see kafka-streaming
Only dbt models on a schedule	dbt’s own scheduler or a minimal job runner

Stay on Airflow when the team already runs it and the need is covered; migration cost is real. Tool-agnostic pipeline design is in data-pipelines.

Gotchas

• Mixing airflow.decorators and airflow.sdk imports in one repo hides which DAGs are migrated; move a folder at a time and enforce with ruff AIR rules.
• {{ ds }} is the logical date. With CronDataIntervalTimetable that is the interval start; with trigger timetables it is the trigger time. Prefer explicit data_interval_start in code.
• start_date in the future means no runs and no error.
• Retrying a task that appends rows or sends emails multiplies the side effect.
• trigger_rule="all_done" cleanup tasks still need their own timeout.
• Top-level Variable.get() turns every parse into a metadata or secrets-backend call.
• Changing dag_id creates a new DAG with no history; the old one stays until deleted.

Output

**DAG:** orders_daily (Airflow 3.3), CronDataIntervalTimetable 02:00 UTC, window = previous day
**Tasks:** extract (overwrite S3 partition) -> load (pool=warehouse, replace partition) -> validate (retries=0)
**Reliability:** retries 3 exp backoff, execution_timeout 45m, dagrun_timeout 3h, DeadlineAlert 90m after queue
**Waits:** S3 sensor deferrable=True, timeout 2h
**Checks:** check_dag_files.py clean (--strict); DagBag test green; parse 0.21 s
**Verified:** `airflow dags test orders_daily 2026-10-18` loaded 48,211 rows; rerun left count unchanged