Airflow DAGs: The Import Time Trap That Stalls Scheduling

One DAG file took 12 seconds to import. The scheduler parses every file on every heartbeat, so that single file delayed scheduling for the whole folder. Workers sat idle while the parser waited on a database handshake.

The culprit was one line at module top level: a DB connection opened at import time. You'll learn why the scheduler executes your top-level code constantly.

We'll dissect DAG anatomy, default_args traps, and the parse loop. Keep top-level pure and scheduling stays boring.

Anatomy of a DAG File

A DAG file has five landmarks: imports, the DAG declaration with schedule and start_date, default_args, tasks, and dependencies. The declaration sets identity: dag_id, tags, description, doc_md. Everything else hangs off it.

Keep the file's job narrow. It describes structure for the parser and defers behavior to task runtime. A reviewer should grasp the pipeline from the dependency chain alone.

Name files after dag_ids. billing_daily.py holds billing_daily. Future you, grepping at midnight, says thanks.

Full declarations also carry dagrun_timeout, max_active_runs, end_date, and render_template_as_native_obj. You'll set schedule=None for manual-only DAGs and schedule='@once' for migrations. Keep dag_id unique — it's the fleet-wide key.

default_args and Why Inheritance Surprises People

default_args supplies defaults every task inherits: owner, retries, retry_delay, email alerts. Task-level arguments override them, which is exactly how you give a flaky API task five retries while the rest keep three.

The trap is sharing one dict object across files. Mutating it anywhere retunes every DAG that references it. Copy, do not share.

Prefer dict(BASE_ARGS) or copy.deepcopy for nested values. Explicit beats clever when six teams share a folder.

import pendulum
from airflow.sdk import dag, task

BASE_ARGS = {
    "owner": "billing-team",
    "retries": 3,
    "retry_delay": pendulum.duration(minutes=5),
}

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args=dict(BASE_ARGS),
    tags=["billing", "foundations"],
    description="Nightly billing aggregation.",
    doc_md="## billing_daily\nIdempotent nightly rollup keyed on invoice_id.",
)
def billing_daily():
    @task
    def extract() -> str:
        return "2026-09-03"

    @task
    def load(partition: str) -> None:
        print(f"loading {partition}")

    load(extract())

billing_daily()

Tags and doc_md for Discoverability

Tags are the UI's search index. Team, domain, and tier tags turn a 500-row DAG list into a filtered view during incidents. description adds the one-line summary; doc_md renders the full runbook page.

Write docs assuming a stranger debugs your DAG at 3 AM. Owner, data source, rerun safety, and who to page belong in doc_md. That page is read more than your code comments.

Audit quarterly. Untagged DAGs are abandoned DAGs; pause or delete them before they confuse the next on-call.

The Parse Loop: What the Scheduler Reads and When

The scheduler loop imports every DAG file, builds a DagBag, and creates runs for due schedules. It repeats on min_file_process_interval cadence. Your module-level code is the loop's inner body.

That is why milliseconds matter. Three hundred milliseconds times 200 files is one minute per loop. Twelve seconds times one file blocks everything behind it.

Watch parse metrics like latency budgets. When the loop falls behind, runs start late with zero worker saturation to explain it.

Tune the loop with min_file_process_interval and watch airflow dags list, airflow tasks list <dag>, and airflow dags show <dag> (graphviz) to prove what the parser saw. Test one date with airflow tasks test <dag> <task> 2015-06-01 and the whole run with airflow dags test <dag> 2026-09-03 — the latter writes no DB state.

Top-Level Code Must Be Side-Effect Free

Side-effect free means importing the file changes nothing in the world: no connections, no queries, no API calls, no file writes. Imports of light modules are fine. Client constructors are not.

Lazy imports inside task functions are the standard escape. Pandas, hooks, and SDK clients load at task runtime on workers, never in the parser. The first task run pays the import once.

Enforce it mechanically. A CI check that greps for Hook( and connect( at module level catches more than code review ever will.

# WRONG: connection opens on every scheduler heartbeat
# db_hook = PostgresHook(postgres_conn_id="billing_db")  # never at top level

import pendulum
from airflow.sdk import dag, task

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["billing"],
)
def billing_clean():
    @task
    def extract() -> list:
        from airflow.providers.postgres.hooks.postgres import PostgresHook

        hook = PostgresHook(postgres_conn_id="billing_db")
        return hook.get_records("SELECT invoice_id FROM invoices LIMIT 100")

    @task
    def load(rows: list) -> None:
        print(f"loaded {len(rows)} rows")

    load(extract())

billing_clean()

is_triggered and the @dag Decorator Style

The @dag decorator turns a function into a DAG factory with schedule, start_date, and tags as arguments. Calling task functions inside wires dependencies automatically through XComArgs. No >> operators needed.

Classic with DAG() blocks still work and dominate legacy codebases. They suit operator-heavy files where explicit bit-shifts read clearly. Both compile to the same graph.

New files should default to decorator style. It composes with TaskFlow, types cleanly, and keeps the dependency chain visible as plain function calls.

import pendulum
from airflow.sdk import dag, task

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["billing"],
    description="Decorator-style billing DAG.",
)
def billing_decorator_style()

# templating inside any operator:
# BashOperator(bash_command="echo {{ ds }} {{ macros.ds_add(ds, 7) }}", task_id="echo_ds")
# use pendulum for tz-aware starts; never datetime.now() as start_date:
    @task(retries=3)
    def extract() -> str:
        return "2026-09-03"

    @task
    def transform(partition: str) -> str:
        return partition.upper()

    @task
    def load(clean: str) -> None:
        print(f"loading {clean}")

    load(transform(extract()))

billing_decorator_style()

# templating inside any operator:
# BashOperator(bash_command="echo {{ ds }} {{ macros.ds_add(ds, 7) }}", task_id="echo_ds")
# use pendulum for tz-aware starts; never datetime.now() as start_date