Skip to content
Toolforge

3. Writing DAGs

A DAG file is a Python file that, when imported, produces DAG objects. That sentence contains the single most consequential fact about writing Airflow: your DAG file is imported constantly, and everything at its top level runs every time.

The two ways to write a DAG

Section titled “The two ways to write a DAG”

TaskFlow (modern, use this):

from airflow.sdk import dag, task
import pendulum


@dag(
    schedule="0 2 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 2, "retry_delay": pendulum.duration(minutes=5)},
    tags=["sales", "daily"],
)
def daily_sales():


    @task
    def extract(**context):
        return {"date": context["ds"], "rows": 4211}


    @task
    def transform(payload: dict) -> int:
        return payload["rows"] * 2


    @task
    def load(total: int):
        print(f"loading {total}")


    load(transform(extract()))


daily_sales()

Classic operators (still everywhere, still correct for non-Python work):

from airflow import DAG
from airflow.providers.standard.operators.bash import BashOperator


with DAG("daily_sales", schedule="0 2 * * *",
         start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
         catchup=False) as dag:


    extract = BashOperator(task_id="extract", bash_command="./extract.sh {{ ds }}")
    load    = BashOperator(task_id="load",    bash_command="./load.sh {{ ds }}")


    extract >> load

They mix freely in one DAG. Use TaskFlow for Python, operators for everything else, and do not rewrite working operator DAGs for the sake of it.

The top-level code problem

Section titled “The top-level code problem”

The most important thing in this chapter.

A DAG file is parsed every min_file_process_interval seconds — by default 30 — for the lifetime of the installation. Anything at module level runs on every one of those parses.

# ✗ CATASTROPHIC
import pandas as pd
import requests


config = requests.get("https://config.internal/dag-config").json()   # ← every 30s
df = pd.read_csv("s3://bucket/lookup.csv")                            # ← every 30s
conn = Variable.get("db_password")                                    # ← every 30s
                                                                       #   a DB query!


@dag(...)
def my_dag():
    @task
    def work():
        ...

That file makes an HTTP call, an S3 read and a database query every thirty seconds, forever, on the DAG processor and on every worker that parses it to run a task. With a hundred such DAGs the processor never finishes a cycle, scheduling falls behind, and the symptom is “Airflow is slow” with nothing in the logs.

# ✓ CORRECT — everything expensive is INSIDE a task
@dag(...)
def my_dag():
    @task
    def work():
        import pandas as pd            # import inside, too
        cfg = requests.get("https://config.internal/dag-config").json()
        df = pd.read_csv("s3://bucket/lookup.csv")
        ...

The rules:

Nothing at module level may do I/O. No HTTP, no database, no file reads, no Variable.get(), no Connection.get().

Import heavy libraries inside tasks. import pandas at the top of every DAG file costs a second per parse per file.

Never call datetime.now() at module level. A start_date that moves every parse produces a DAG whose schedule is undefined.

A DAG file should parse in well under a second. Measure it:

Terminal window
airflow dags list-import-errors
airflow dags report            # parse duration per file
python -c "import time; s=time.time(); import my_dag; print(time.time()-s)"

Anything over a second deserves attention; over five seconds is a problem you already have.

There is a mitigation worth knowing about — @task decorated functions and operator arguments are lazily evaluated, and Airflow 2.4+ can skip re-parsing unchanged files — but neither saves you from module-level I/O. Move it into tasks.

Templating

Section titled “Templating”

Airflow renders Jinja in templated operator fields at run time, which is how a task learns which date it is running for:

BashOperator(
    task_id="extract",
    bash_command="./extract.sh --date {{ ds }} --run {{ run_id }}",
)

The variables you will actually use:

Variable Value
{{ ds }} logical date, 2026-08-30
{{ ds_nodash }} 20260830
{{ data_interval_start }} start of the interval this run covers
{{ data_interval_end }} end of it — usually what you want
{{ logical_date }} the run’s logical timestamp
{{ run_id }} unique identifier for the run
{{ ti }} the task instance object
{{ params.x }} a DAG parameter
{{ var.value.x }} an Airflow Variable — rendered at run time, safe
{{ conn.my_conn.host }} a connection field

That var.value row is the safe way to use a Variable. Variable.get() at module level is a database query on every parse; {{ var.value.x }} is fetched when the task runs.

Chapter 4 is entirely about ds versus data_interval_end, because the difference is the single most confusing thing in Airflow.

In TaskFlow, the context arrives as keyword arguments:

@task
def extract(**context):
    start = context["data_interval_start"]
    end   = context["data_interval_end"]

or explicitly:

from airflow.sdk import get_current_context


@task
def extract():
    ctx = get_current_context()

Dynamic DAG generation

Section titled “Dynamic DAG generation”

Twenty near-identical pipelines should not be twenty near-identical files.

dags/generated.py
import yaml, pathlib
from airflow.sdk import dag, task


CONFIG = pathlib.Path(__file__).parent / "configs"


def build(name: str, cfg: dict):
    @dag(dag_id=f"ingest_{name}", schedule=cfg["schedule"],
         start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
         catchup=False)
    def _dag():
        @task
        def ingest():
            run_ingest(cfg["source"], cfg["target"])
        ingest()
    return _dag()


for path in CONFIG.glob("*.yaml"):
    cfg = yaml.safe_load(path.read_text())     # local file read: acceptable
    globals()[f"ingest_{path.stem}"] = build(path.stem, cfg)

Two rules make this safe.

The DAG must be assigned into globals(). Airflow finds DAGs by scanning module globals; a DAG object created inside a function and not assigned is invisible.

The configuration source must be cheap and local. Reading YAML files next to the DAG is fine — it happens on every parse but costs microseconds. Reading configuration from a database or an API is the top-level code problem in a costume, and it is how people accidentally build the pathological version of this pattern.

If configuration genuinely lives in a remote system, materialise it into files in CI and commit them. The DAG file then reads local files, and the expensive fetch happens once at build time rather than every thirty seconds forever.

Dynamic task mapping

Section titled “Dynamic task mapping”

Different problem: the number of tasks is not known until run time.

@task
def list_files() -> list[str]:
    return s3_list("bucket/incoming/")        # 1 file or 400, unknown


@task
def process(path: str):
    handle(path)


process.expand(path=list_files())

.expand() creates one task instance per element at run time. The graph’s shape is still fixed — one mapped task — but its width is dynamic. This is Airflow’s answer to “loop over a list”, and it is the right one.

With constants alongside the varying argument:

process.partial(bucket="prod-data").expand(path=list_files())

The limit worth knowing: max_map_length defaults to 1024. Expanding over ten thousand files fails. And each mapped instance is a row in task_instance — a mapped task over 500 items, run hourly, is 12,000 rows a day from one task. Chapter 10 explains why that matters.

The pattern that avoids it: map over batches, not items. One hundred tasks each handling fifty files beats five thousand tasks handling one.

Branching

Section titled “Branching”
from airflow.sdk import task


@task.branch
def choose(**context):
    if context["data_interval_start"].day_of_week == 0:
        return "weekly_report"
    return "skip_it"

The branch task returns the task_id (or ids) to follow; everything else downstream is marked skipped. Which produces the classic surprise: a join task after a branch does not run, because its default trigger rule requires all upstream tasks to have succeeded and some were skipped. The fix is trigger_rule="none_failed_min_one_success", and chapter 8 covers trigger rules properly.

default_args

Section titled “default_args”

Applied to every task in the DAG:

default_args = {
    "owner": "data-platform",
    "retries": 2,
    "retry_delay": pendulum.duration(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": pendulum.duration(hours=1),
    "execution_timeout": pendulum.duration(hours=2),
    "on_failure_callback": alert_slack,
}

execution_timeout deserves to be in every one of these. Without it a hung task holds a worker slot indefinitely, and a handful of them can consume your entire pool. It is the single most valuable default argument and it is not set by default.

retry_exponential_backoff is the other one worth having: retrying a failing external API every five minutes forever is how you get rate-limited.

Testing DAGs

Section titled “Testing DAGs”

Underdone almost everywhere, and cheap to fix.

tests/test_dags.py
from airflow.models import DagBag


def test_no_import_errors():
    bag = DagBag(include_examples=False)
    assert not bag.import_errors, bag.import_errors


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

That first test alone catches most breakage, and it belongs in CI on every pull request. A DAG with an import error does not appear in Airflow at all — it does not fail loudly, it is simply absent, which is a much worse failure than a red task.

To run one task locally:

Terminal window
airflow tasks test daily_sales extract 2026-08-30

tasks test runs a single task in-process, with real context, without writing to the metadata database — nothing is recorded, no state changes, no downstream effects. It is the right way to iterate on task code.

Terminal window
airflow dags test daily_sales 2026-08-30     # whole DAG, still no DB state

What to take from this chapter

Section titled “What to take from this chapter”
  • Use TaskFlow (@dag, @task) for Python, classic operators for everything else. They mix freely.
  • The top-level code rule is the most important thing in this chapter. DAG files are parsed every ~30 seconds forever, on the processor and on workers. No I/O at module level — no HTTP, no database, no Variable.get(), no datetime.now(). Import heavy libraries inside tasks.
  • Target well under a second per DAG file; measure with airflow dags report.
  • Use {{ var.value.x }} in templates rather than Variable.get() at module level — it is fetched at run time.
  • Dynamic DAG generation requires assigning into globals() and reading only cheap local configuration. Materialise remote config into files in CI.
  • Dynamic task mapping (.expand()) is the right way to loop. max_map_length is 1024, and every mapped instance is a database row — map over batches, not items.
  • After a branch, unfollowed tasks are skipped, so a downstream join needs trigger_rule="none_failed_min_one_success".
  • Put execution_timeout in default_args. A hung task holds a worker slot forever, and it is not set by default.
  • Test for import errors in CI. A DAG that fails to import does not appear at all — silence, not a red task. Use airflow tasks test to iterate without touching the database.

Next: scheduling — the data interval, why ds is not today’s date, catchup, and the naming that has confused everyone who has ever used Airflow.

Written by Sepuri Sai Krishna

© 2026CC BY 4.0Source