Airflow Dags

Writing Airflow 3 DAGs

Airflow 3 DAGs are Python code that define data pipeline workflows. Every DAG must follow three principles: atomicity (each task does one thing), idempotency (same input = same output on rerun), and modularity (reusable functions and operators, DRY).

Project Structure

project/
  dags/                    # One .py file per DAG, filename = dag_id
  include/                 # Support code (NOT parsed by scheduler)
    sql/
    python_functions/
    custom_operators/
    custom_hooks/
  tests/
    dag_validation/        # DagBag-based structural checks
    unit_tests/            # Custom code with mocked dependencies
    integration_tests/     # Real external systems, no mocking
  plugins/
    cluster_policies/      # Environment-level enforcement (@hookimpl)

Quick Decision Guide

Critical Rules

These rules are non-negotiable. Violating any of them causes production failures.

1. NO Top-Level Code

DAG files are parsed by the scheduler every 30 seconds. Top-level code executes on EVERY parse.

# BAD - executes every 30 seconds during parsing
config = requests.get("https://config-api/settings").json()  # API hammered
data = pd.read_csv("/data/input.csv")                        # I/O on every parse
now = datetime.now()                                          # Different every parse

# GOOD - executes only when task runs
@task()
def fetch_config():
    return requests.get("https://config-api/settings").json()

Allowed at top level: imports, constants, @dag/@task decorators, datetime(2024, 1, 1) literals.

2. Use Airflow Hooks, Not Raw Clients

Always use Airflow hooks (S3Hook, HttpHook, PostgresHook) instead of raw clients (boto3, requests). Hooks use Airflow connections for credential management.

# BAD - hardcoded/env credentials, no connection management
import boto3
s3 = boto3.client("s3")

# GOOD - credentials from Airflow connection
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
hook = S3Hook(aws_conn_id="aws_default")

3. Credentials at Task Time Only

Never fetch credentials during DAG definition. Use Airflow connections or secrets backends.

4. Each Task = One Atomic Unit

One task does one thing: extract OR transform OR load. Never combine. This enables partial reruns and clear observability.

5. Idempotent Tasks

Same input must produce same output. Use partitioning with Airflow context variables ({{ ds }}, {{ data_interval_start }}). Overwrite output, never append.

6. Always Set Retries

default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=2),
    "retry_exponential_backoff": True,
}

7. Always Set Execution Timeout

@task(execution_timeout=timedelta(minutes=30))
def my_task():
    ...

8. Use Deferrable Operators for Long Waits

For tasks waiting >1 minute on external systems, use deferrable operators to release worker slots.

# Set globally: AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True
# Or per-sensor: mode="reschedule" for sensors with long waits

9. One DAG Per File

Filename should match dag_id. All support code goes in include/.

TaskFlow DAG Template

Complete example applying all critical rules:

"""ETL pipeline: API -> Transform -> S3."""
from __future__ import annotations

import json
from datetime import datetime, timedelta

from airflow.decorators import dag, task
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.http.hooks.http import HttpHook

default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=2),
    "retry_exponential_backoff": True,
    "execution_timeout": timedelta(minutes=30),
}


@dag(
    dag_id="api_to_s3_etl",
    start_date=datetime(2024, 1, 1),
    schedule="@hourly",
    catchup=False,
    max_active_runs=1,
    max_consecutive_failed_dag_runs=5,
    default_args=default_args,
    tags=["etl", "api"],
    doc_md=__doc__,
)
def api_to_s3_etl():

    @task()
    def extract(**context) -> list[dict]:
        hook = HttpHook(http_conn_id="api_default", method="GET")
        response = hook.run(endpoint="/data")
        return response.json()

    @task()
    def transform(records: list[dict]) -> list[dict]:
        return [r for r in records if r.get("status") == "active"]

    @task()
    def load(records: list[dict], **context) -> None:
        ds = context["ds"]
        hook = S3Hook(aws_conn_id="aws_default")
        hook.load_string(
            string_data=json.dumps(records),
            key=f"hourly/{ds}/data.json",
            bucket_name="my-data-lake",
            replace=True,
        )

    data = extract()
    filtered = transform(data)
    load(filtered)


api_to_s3_etl()

if __name__ == "__main__":
    api_to_s3_etl().test()

Key patterns in this template:

• @dag/@task decorators (TaskFlow API) — not with DAG() context manager
• Hooks for external systems (HttpHook, S3Hook) — not raw requests/boto3
• Partitioned S3 key using {{ ds }} context — idempotent
• default_args with retries, backoff, timeout
• max_consecutive_failed_dag_runs=5 — auto-pauses DAG after 5 consecutive failures
• if __name__ block for dag.test() development testing
• doc_md for UI documentation
• No top-level computation

Common Mistakes

Reference Files

Deep-dive documentation organized by topic. Claude loads these on-demand:

• DAG Authoring — TaskFlow API, Traditional syntax, Asset-oriented approach, when to use each
• Scheduling — Cron, assets, event-driven, timetables, catchup, backfill
• Dependencies — chain(), >>, trigger rules, branching, task groups
• Dynamic Tasks — .expand(), .partial(), dynamic task groups, dag-factory
• Data Passing — XCom, custom backends, Object Storage backend
• Operators — Operators, sensors, deferrable operators, hooks, providers
• Connections — Connection types, secrets backends, env vars
• ETL/ELT Patterns — 11 practical DAG examples: classic ETL, ELT, incremental, asset-oriented, dynamic, fan-in, dbt, CDC, factory, ETLT hybrid, asset-sequence
• Data Quality — Quality gates, temp-table swap, SQLTableCheckOperator, stopping vs warning checks
• Testing — 5-layer testing: dag.test(), validation, unit, integration, policies
• Production — DAG versioning, bundles, scaling, callbacks, debugging
• Migration — Airflow 2→3 breaking changes, parameter renames, Ruff AIR30 linting