Airflow in Production: Testable DAGs, Sensors, and SLAs

Airflow is not a scheduler. A scheduler runs commands on a clock. Airflow is an orchestrator: it manages dependencies between tasks, retries failures with configurable policy, tracks state across historical runs, and surfaces the operational view of your pipelines through a UI that, when used well, tells you the health of your data platform at a glance. The distinction matters because the patterns that make a scheduled cron job reliable are not the same patterns that make an Airflow DAG reliable.

DAGs that can be tested

The first property I require of every DAG is that it can be imported without error and its structure can be validated without running a single task. Airflow re-imports every DAG file on a configurable interval, and a DAG with an import error does not appear in the UI, produces no useful error message to the user, and silently stops scheduling. An import test catches these issues in the CI pipeline, before deployment.

# tests/test_dag_integrity.py
import importlib
import os
from pathlib import Path

import pytest
from airflow.models import DagBag

DAGS_DIR = Path(__file__).parent.parent / "dags"

def test_no_import_errors():
    dag_bag = DagBag(dag_folder=str(DAGS_DIR), include_examples=False)
    assert dag_bag.import_errors == {}, (
        f"DAG import errors found:\n"
        + "\n".join(f"  {k}: {v}" for k, v in dag_bag.import_errors.items())
    )

def test_dag_structure():
    dag_bag = DagBag(dag_folder=str(DAGS_DIR), include_examples=False)
    for dag_id, dag in dag_bag.dags.items():
        # Every DAG must have an owner that is not the default
        assert dag.owner != "airflow", f"{dag_id}: owner is default 'airflow'"
        # Every DAG must have a description
        assert dag.description, f"{dag_id}: missing description"
        # No task should depend on itself
        assert dag.test_cycle() is None, f"{dag_id}: cycle detected"

Beyond import tests, the transformation logic inside tasks should be factored into pure Python functions that can be unit-tested independently of Airflow. A task that calls a function is testable. A task that contains the logic inline is only testable by running Airflow.

# transforms/orders.py: pure functions, no Airflow imports
def compute_order_revenue(orders: list[dict], exchange_rates: dict[str, float]) -> list[dict]:
    return [
        {**order, "revenue_usd": order["amount"] * exchange_rates.get(order["currency"], 1.0)}
        for order in orders
        if order["status"] == "completed"
    ]

# dags/orders_pipeline.py: Airflow wrapper calls the function
from airflow.decorators import task
from transforms.orders import compute_order_revenue

@task
def transform_orders(orders: list[dict], exchange_rates: dict) -> list[dict]:
    return compute_order_revenue(orders, exchange_rates)

# tests/test_transforms.py: tests the function, not the task
def test_compute_order_revenue_filters_non_completed():
    orders = [
        {"order_id": "A", "amount": 100, "currency": "BRL", "status": "completed"},
        {"order_id": "B", "amount": 50,  "currency": "BRL", "status": "cancelled"},
    ]
    rates = {"BRL": 0.20}
    result = compute_order_revenue(orders, rates)
    assert len(result) == 1
    assert result[0]["revenue_usd"] == pytest.approx(20.0)

Sensors: waiting for what is not yet there

A sensor is a task whose job is to wait for a condition to be true before allowing downstream tasks to proceed. The condition might be the presence of a file in S3, a row in a database table, or the completion of an external process. Without sensors, downstream tasks either fail immediately (if the upstream data is not ready) or run on stale data (if they do not check).

# Wait for the upstream orders export to land in S3 before processing
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.operators.python import PythonOperator

with DAG("orders_pipeline", schedule="0 6 * * *", ...):
    wait_for_export = S3KeySensor(
        task_id="wait_for_orders_export",
        bucket_name="data-lake-raw",
        bucket_key="exports/orders/{{ ds }}/orders_{{ ds }}.parquet",
        aws_conn_id="aws_default",
        timeout=3600,           # fail after 1 hour of waiting
        poke_interval=300,      # check every 5 minutes
        mode="reschedule",      # release the worker slot while waiting
    )

    process_orders = PythonOperator(
        task_id="process_orders",
        python_callable=run_orders_pipeline,
    )

    wait_for_export >> process_orders

The mode="reschedule" setting is important in production. In poke mode, the sensor holds a worker slot for its entire duration, which can exhaust the worker pool if many sensors are waiting simultaneously. In reschedule mode, the sensor releases the slot between checks and re-acquires it only when rescheduled for the next poke. This costs a small amount of scheduling overhead and provides a significant reduction in worker slot consumption.

SLAs and miss callbacks

An SLA in Airflow is the maximum elapsed time from a DAG's scheduled start time to the completion of a specific task. When the task does not complete within the SLA, Airflow calls the sla_miss_callback function. This is the hook that connects pipeline latency to operational response.

from airflow import DAG
from airflow.models import TaskInstance, SlaMiss
from datetime import timedelta
import smtplib

def notify_sla_miss(dag, task_list, blocking_task_list, slas, blocking_tis):
    message_lines = [f"SLA miss in DAG: {dag.dag_id}"]
    for sla in slas:
        message_lines.append(
            f"  Task '{sla.task_id}' missed SLA. "
            f"Scheduled: {sla.execution_date}, SLA: {sla.sla}"
        )
    message_lines.append("\nBlocking tasks (still running or failed):")
    for ti in blocking_tis:
        message_lines.append(f"  {ti.task_id}: {ti.state}")
    message_lines.append("\nRunbook: https://wiki.internal/pipelines/orders/runbook")

    send_alert(subject=f"SLA Miss: {dag.dag_id}", body="\n".join(message_lines))

with DAG(
    "orders_pipeline",
    schedule="0 6 * * *",
    sla_miss_callback=notify_sla_miss,
    default_args={
        "sla": timedelta(hours=2),  # each task must complete within 2h of DAG start
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "retry_exponential_backoff": True,
    },
) as dag:
    ...

What I learned orchestrating real workloads

The most consistent production failure mode I encountered was DAGs that were too large. A single DAG with 40 tasks, branching logic, and multiple upstream dependencies is difficult to debug when it fails because the failure could be anywhere. Small DAGs (5 to 15 tasks, one responsibility per DAG) are faster to diagnose, easier to test, and produce cleaner dependency graphs in the UI.

Dynamic task generation (building task lists at DAG parse time based on database queries or file system state) creates DAGs that behave differently on every parse, are hard to test, and produce unpredictable UI states. The pattern I apply instead is static DAG structure with dynamic parameters: the DAG structure is fixed, and the parameters (which date range to process, which files to load) are passed at runtime through Airflow variables or the triggered run configuration.

# Bad: dynamic task generation at parse time
for table in db.fetch_table_list():  # runs on every DAG parse
    PythonOperator(task_id=f"load_{table}", ...)

# Good: static structure with runtime parameterization
@task
def load_tables(tables: list[str]) -> None:
    for table in tables:
        load_table(table)

# Tables list passed at trigger time or read inside the task, not at parse time

The task that reads the list of tables inside its execution context runs once per DAG run, uses current data, and produces the same DAG structure on every parse. The scheduler parses it cleanly, the UI renders it consistently, and tests can run against it without a live database connection.