---
name: airflow-dag-conventions
description: "Structures an Apache Airflow DAG the production-safe way — no heavy work at import/parse time, tasks that are idempotent and re-run safe, backfill turned off by default, explicit retries and a task timeout, sensors bounded by a timeout and freed while waiting, and only small pointers passed between tasks instead of large data. Use when writing, generating, or reviewing an Airflow DAG or its tasks. Do NOT use for choosing an orchestrator, tuning warehouse SQL, or planning a historical reprocess."
metadata:
  decimalai-skill-id: "59281485-fcd1-4cc9-9b43-c1cf71d87368"
  decimalai-version-id: "6f0bcf3e-f255-4760-99bd-ec53cc28bf17"
  decimalai-version: "1"
  decimalai-body-sha256: "d498710f0093c4945408e29230a92b19776652f7af4be088fe2b67d9c635b77b"
  decimalai-name: "airflow-dag-conventions"
  decimalai-format: "1"
  decimalai-invocation: "model"
  decimalai-category: "data"
  decimalai-source: "https://app.decimal.ai/s/airflow-dag-conventions@1/SKILL.md"
---
# Airflow DAG Conventions

Writes and reviews an **Apache Airflow DAG** so it survives production: the scheduler that parses it thousands of times a day, the retries and backfills that re-run its tasks, and the queue slots its sensors would otherwise hold hostage. A DAG that "works" the first time you trigger it can still take down the scheduler, silently double-write, or hang a worker forever. These conventions are the difference.

The base model knows Airflow's building blocks but writes a naive DAG by default. Asked to "write a DAG that loads yesterday's orders," it will happily run a database query at module top level, leave the default that replays all of history on the first unpause, append rows a retry will duplicate, omit retries and timeouts, use a sensor that blocks a worker slot with no deadline, and return a whole DataFrame from a task. Each of those is a production incident. Apply the six conventions below to every DAG you write or review.

## When to Activate

Activate when the user wants to:
- Write or generate an Airflow DAG, a task, or a custom operator
- Review or fix an existing DAG (slow scheduler, duplicate rows on retry, a stuck task, a backfill that ran away)
- Convert a script or notebook into a scheduled Airflow pipeline
- Add a sensor, a retry policy, or cross-task data passing to a DAG

Do **not** activate for choosing between orchestrators, tuning the SQL a task runs, or planning a one-off historical reprocess (that is `backfill-safety`).

## The Six Conventions

Check every DAG against all six. A violation of any one is a finding, with the safe form named.

### 1. Nothing heavy at parse / top level
The DAG file is **imported by the scheduler on every parse cycle** — every few seconds, forever — just to discover its structure. Any code at module top level (outside a task/operator) runs on every one of those parses. So a database query, an API call, a file read, a `pandas` load, or a `Variable.get()` at the top of the file hammers that resource thousands of times a day and slows the whole scheduler's parse loop for every DAG.

Keep the top level to cheap structure only: imports, the `DAG`/`@dag` definition, and operator declarations. Push all real work — connections, queries, computation, credential fetches — **inside** the task function (or the operator's `execute`), where it runs only when the task runs. Building the list of tasks from a small static config is fine; fetching that config from a database at parse time is not.

### 2. Tasks are idempotent — a re-run changes nothing
Airflow re-runs tasks constantly: retries, manual clears, backfills, catchup. Each run for a given logical date must produce the **same** result whether it runs once or five times. Two rules follow:
- **Partition by the run's logical date, and overwrite that partition** — `DELETE`/`INSERT OVERWRITE`/`MERGE` scoped to `{{ ds }}` (the run's date), not a bare `INSERT ... APPEND`. A plain append doubles a partition's rows on the first retry.
- **Derive time from the run, not the wall clock.** Use the execution/logical date (`{{ ds }}`, `data_interval_start`) to pick the window. `datetime.now()` / `date.today()` inside a task makes a retry or a backfill process a *different* window than the original run — the output no longer matches the date it is filed under.

### 3. `catchup=False` unless a backfill is intended
Airflow's default is `catchup=True`. Unpause a DAG with a `start_date` months in the past and it will immediately schedule **every missed interval** since then — a stampede of runs that overloads the source and floods the warehouse. Set `catchup=False` on the DAG by default so it runs from now forward; only turn catchup on (or run an explicit bounded backfill) when replaying history is the actual goal. Also give the DAG a **fixed** `start_date` (a literal date), never `datetime.now()` — a dynamic start date makes the first interval never elapse and the DAG never runs.

### 4. Explicit retries and a task timeout
Do not ship a task on defaults. Set, in `default_args` or per task:
- `retries` (e.g. 2–3) and a `retry_delay` — transient source/network failures are normal and should self-heal, not page someone.
- `execution_timeout` — a hung query or a wedged API call must be killed, not left running until it blocks the pool. A task with no timeout can hang indefinitely and starve every other task of slots.
- An **SLA or a max-runtime expectation** where lateness matters, so a task that is merely slow (not failed) still surfaces.

Silent-forever is the failure mode these prevent: without a timeout a stuck task looks "running" for hours; without retries a one-second network blip fails the whole run.

### 5. Sensors are bounded and yield while waiting
A sensor waits for something (a file, a partition, another DAG). Two defaults make sensors dangerous:
- **No timeout** — the default sensor waits *forever*. Always set a `timeout` so a dependency that never arrives fails the task instead of hanging until someone notices.
- **`poke` mode holds a worker slot the entire time it waits.** A few sensors in default poke mode can occupy every slot in the pool and deadlock the DAG (nothing else can run, so the thing being waited on never gets produced). Use `mode='reschedule'` (or a deferrable/async sensor) so the sensor **releases its slot between checks** and only reoccupies one to poke. Also set a sane `poke_interval` — don't check a slow dependency every few seconds.

### 6. Only small pointers through XCom — never large data
XCom values are **stored in Airflow's metadata database** and pulled back through it. It is a channel for small scalars: an id, a row count, a file path, a partition name. Returning a DataFrame, a full query result, a big JSON blob, or a file's contents from a task pushes megabytes into the metadata DB — bloating it, slowing every scheduler query against it, and often hitting a hard size limit outright.

Pass a **reference**, not the payload: write the data to object storage / a table / a temp file and push only its **path or key** through XCom; the downstream task reads it back from that location. If two tasks need to share a large dataset, they share a storage location, not an XCom value.

## Output

When reviewing a DAG, present findings as a table, then a verdict.

| # | Severity | Convention | Finding | Safe form |
|---|----------|-----------|---------|-----------|
| 1 | 🔴 Block | Parse-time work | A warehouse query runs at module top level, so it fires on every scheduler parse | Move the query inside the task function |
| 2 | 🔴 Block | Idempotency | Task appends rows, so a retry duplicates the partition | Overwrite the `{{ ds }}` partition instead of appending |

**Verdict:** SHIP / FIX FIRST / BLOCK — one line naming the biggest reason.

When *writing* a DAG, produce one that already satisfies all six, and call out in a sentence which conventions drove the non-obvious choices (catchup, reschedule sensor, XCom pointer).

## Examples

### Input
> Write me an Airflow DAG that every morning pulls yesterday's rows from our `orders` Postgres table, aggregates a daily total per store, and writes the result to a `daily_store_totals` table. Keep it simple.

### Expected Output
A DAG whose module top level holds only imports and the DAG/task declarations — the Postgres connection and query live inside the task, not at parse time. `catchup=False`, a literal `start_date`. The task derives its window from the run's logical date (`{{ ds }}`), and writes by deleting-then-inserting (or upserting) that date's partition of `daily_store_totals`, so a retry cannot duplicate. `default_args` set `retries`, `retry_delay`, and `execution_timeout`. If it returns anything to a downstream task, it returns the partition key, not the rows. A one-line note explains the catchup and idempotency choices.

### Input
> Review this DAG before I turn it on:
> ```python
> conn = psycopg2.connect(DB_URL)
> rows = conn.execute("SELECT * FROM events").fetchall()   # at module level
> dag = DAG("etl", start_date=datetime.now())
> def load(**c):
>     df = extract()          # returns a big DataFrame
>     return df               # pushed to XCom
> wait = FileSensor(task_id="wait", filepath="/data/in.csv")  # default poke, no timeout
> ```

### Expected Output

| # | Severity | Convention | Finding | Safe form |
|---|----------|-----------|---------|-----------|
| 1 | 🔴 Block | Parse-time work | The connection and `SELECT` run at module top level, so they hit Postgres on every scheduler parse | Move the connect + query inside the task |
| 2 | 🔴 Block | Dynamic start_date / catchup | `start_date=datetime.now()` plus default `catchup=True`; the interval never closes and/or history stampedes | Use a fixed `start_date` and set `catchup=False` |
| 3 | 🔴 Block | XCom size | `load` returns a whole DataFrame into XCom, bloating the metadata DB | Write the frame to storage; return only its path |
| 4 | 🔴 Block | Sensor bound | `FileSensor` has no `timeout` and default poke mode holds a worker slot indefinitely | Add a `timeout` and `mode='reschedule'` |
| 5 | 🟡 Warn | Retries / timeout | No `retries` or `execution_timeout` on the tasks | Set both in `default_args` |

**Verdict:** BLOCK — the parse-time query and the DataFrame-through-XCom will hurt the scheduler and the metadata DB the moment this is unpaused.

## Edge Cases

- **Small static config at parse time** (a hard-coded list of table names used to build tasks) is fine — the "no parse-time work" rule targets I/O and computation, not cheap literals. Don't flag it.
- **Dynamic task mapping** still must not do the expensive fan-out computation at parse time; map over a small static input or a cheap source, not a full query.
- **A genuinely intended backfill** is the one case `catchup=True` (or an explicit bounded backfill run) is correct — but it still requires the idempotent, date-partitioned writes of convention 2, or the replay duplicates data.
- **Tiny XCom values** — an id, a count, a path — are exactly what XCom is for; don't push users toward external storage for a scalar.
- **Deferrable operators** are an even better answer than `reschedule` for long waits (they free the slot entirely via the triggerer); accept them as satisfying convention 5.
- **A short, fast, self-contained task** may not need an SLA — retries and a timeout still apply, but don't demand an SLA where lateness is meaningless.

## Evaluation Criteria

A good execution of this skill should:
- [ ] Keep all I/O and computation out of module/parse-time scope and inside tasks
- [ ] Make each task idempotent — date-partitioned overwrite and time derived from the logical date, not the wall clock
- [ ] Set `catchup=False` by default and use a fixed `start_date`
- [ ] Give tasks explicit `retries`/`retry_delay` and an `execution_timeout`
- [ ] Bound sensors with a `timeout` and free the worker slot while waiting (`reschedule`/deferrable)
- [ ] Pass only small pointers through XCom, never large datasets
- [ ] Not over-flag benign cases (small static parse-time config, scalar XCom, intended backfills)