Airflow DAGs skill for Codex

Done well means: re-running any task for any interval produces the same result, the DAG file parses in well under a second with no side effects, failures retry and then alert, and backfills are bounded and confirmed before they touch shared systems.

Before writing code

1. Read AGENTS.md, the dags/ folder layout, requirements.txt or the image definition, and any shared helpers (dags/common/, plugins).
2. airflow version. Version decides imports and defaults:

   Topic	Airflow 2.x	Airflow 3.x
   Authoring imports	from airflow.decorators import dag, task	from airflow.sdk import dag, task, Asset
   Schedule arg	schedule (2.4+), schedule_interval deprecated	schedule only
   catchup default	True	False
   Data-aware scheduling	Dataset	Asset
   Backfill	airflow dags backfill -s ... -e ...	airflow backfill create ... (scheduler-managed)
   Date context	execution_date deprecated	execution_date removed; logical_date can be None for manual or asset-triggered runs
   Task DB access	Allowed (discouraged)	Tasks cannot query the metadata DB directly
   Bash and Python operators live in airflow.providers.standard on 3.x.

3. Find out what the DAG writes to and whether those writes are idempotent. That drives the design.

Design rules

• One DAG per pipeline with a clear owner and schedule; one task per unit you want to retry independently.
• Each task processes exactly one data interval: read [data_interval_start, data_interval_end), write to a location keyed on that interval, and overwrite (or MERGE) rather than append.
• Tasks pass references through XCom (paths, table names, row counts), never DataFrames or large payloads.
• Heavy compute runs elsewhere (warehouse, Spark, Kubernetes pod); Airflow orchestrates.
• No work at module top level: no API calls, DB queries, Variable.get, or file listings outside tasks. The scheduler parses every DAG file continuously.

TaskFlow template (Airflow 3)

from __future__ import annotations

from datetime import timedelta

import pendulum
from airflow.sdk import Asset, dag, task

orders_daily = Asset("s3://lake/orders_daily/")

@dag(
    schedule="0 3 * * *",                       # cron in the DAG timezone (UTC here)
    start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
    catchup=False,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5),
        "retry_exponential_backoff": True,
        "execution_timeout": timedelta(hours=1),
    },
    tags=["orders", "daily"],
)
def orders_daily_pipeline():

    @task
    def extract(data_interval_start=None, data_interval_end=None) -> str:
        path = f"s3://lake/raw/orders/date={data_interval_start:%Y-%m-%d}/"
        export_orders(since=data_interval_start, until=data_interval_end, dest=path)  # overwrites
        return path

    @task(pool="warehouse", outlets=[orders_daily])
    def load(path: str, data_interval_start=None) -> int:
        # delete-then-insert for the interval keeps reruns idempotent
        return merge_partition(path, partition=data_interval_start.date())

    @task
    def check(rows: int) -> None:
        if rows == 0:
            raise ValueError("no orders loaded for interval")

    check(load(extract()))

orders_daily_pipeline()

if __name__ == "__main__":
    orders_daily_pipeline().test()   # runs locally without a scheduler

Context values (data_interval_start, ti, params) are injected when declared as keyword arguments with a default of None; get_current_context() also works inside a task.

Choosing building blocks

Need	Use
Fan out over a list known only at runtime	Dynamic task mapping: process.expand(path=list_files()), cap with max_active_tis_per_dag
Wait for a file or external condition	Deferrable sensor or mode="reschedule" with timeout; never long poke mode
Start DAG B when DAG A updated data	Assets: A’s task has outlets=[asset], B has schedule=[asset]
Protect a rate-limited system	pool="name" with a slot count matching its limit
Cleanup that must run after failures	trigger_rule="all_done"
Join after a branch	trigger_rule="none_failed_min_one_success"
Group related tasks visually	@task_group
Credentials	Airflow Connections via provider hooks with conn_id; never literals in DAG files

Backfills and reruns

Confirm with the user before any backfill or bulk clear on a shared environment, stating the date range, number of runs and the systems written to.

# Airflow 3: scheduler-managed backfill
airflow backfill create --dag-id orders_daily_pipeline \
    --from-date 2025-08-01 --to-date 2025-08-31 \
    --reprocess-behavior failed --max-active-runs 2

# Airflow 2.x
airflow dags backfill -s 2025-08-01 -e 2025-08-31 orders_daily_pipeline

--reprocess-behavior: none skips dates that already have a run, failed reruns only failed ones, completed reruns everything that finished. Backfills are only safe if tasks are idempotent; if they append, fix that first.

Testing and local checks

python dags/orders_daily.py                      # executes dag.test() for one run
airflow dags list-import-errors
airflow dags test orders_daily_pipeline 2025-09-01
airflow tasks test orders_daily_pipeline load 2025-09-01   # one task, no state recorded

# tests/test_dag_integrity.py
from airflow.models import DagBag

def test_no_import_errors():
    bag = DagBag(dag_folder="dags", include_examples=False)
    assert bag.import_errors == {}
    for dag_id, dag in bag.dags.items():
        assert dag.default_args.get("retries", 0) >= 1, dag_id
        assert dag.tags, dag_id

Measure parse time with airflow dags report and keep each file well under the dagbag_import_timeout (default 30 s); aim for under one second.

Gotchas

• start_date=datetime.now() or any dynamic start date breaks scheduling. Use a fixed date.
• A run for interval 2025-09-01 starts after the interval ends (2025-09-02 03:00 here). People expect it the same day; explain it when setting schedules.
• Cron without an explicit timezone runs in the configured default, usually UTC; DST shifts apply only with a timezone-aware start_date.
• catchup=True on a DAG with an old start date launches hundreds of runs on unpause.
• Mapped tasks over thousands of items flood the scheduler; batch items into chunks.
• Retrying a non-idempotent task (append inserts, sending emails) duplicates effects. Make outward-facing actions explicit, single-attempt tasks with retries=0, and confirm sends.
• SLA callbacks were removed in Airflow 3.0; use timeouts plus failure callbacks or the newer deadline alerts on versions that have them.

Review checklist

- [ ] Fixed start_date, explicit catchup, max_active_runs set
- [ ] Every task idempotent for its data interval (overwrite / merge, no blind append)
- [ ] retries, retry_delay, execution_timeout on every task
- [ ] No top-level side effects; parse time < 1 s
- [ ] Sensors deferrable or reschedule with timeout
- [ ] Credentials via Connections; secrets not in code or XCom
- [ ] DagBag integrity test passes; dags test run succeeds for one interval

Report template

**DAG:** orders_daily_pipeline (Airflow 3.1, daily 03:00 UTC)
**Tasks:** extract -> load (pool warehouse, 4 slots) -> check; emits Asset s3://lake/orders_daily/
**Idempotency:** partition overwrite keyed on data_interval_start; rerun of 2025-09-01 produced identical row count
**Tests:** DagBag integrity pass; airflow dags test for 2025-09-01 pass (2m 41s)
**Backfill plan (needs your approval):** 2025-08-01..2025-08-31, 31 runs, max 2 active