Production,
diagnostics,
and monitoring

The first gate: can every DAG import?

from airflow.models import DagBag

def test_no_import_errors():
    dag_bag = DagBag()
    assert len(dag_bag.import_errors) == 0

• DagBag scans and imports DAG files like the DAG Processor.
• It records SyntaxError, NameError, missing modules, and top-level exceptions.
• This gate proves importability, not business correctness.

Break a DAG, observe the gate, fix it

from airflow.decorators import dag, task
import nonexistent_module  # Intentionally missing.

@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False)
def broken_import_dag():
    @task
    def run():
        print("This task will never run")
    run()

A minimal deployment workflow

- name: Lint
  run: ruff check dags/

- name: Import test
  run: pytest tests/test_dag_integrity.py -v

- name: Deploy DAGs
  if: success()
  run: gsutil -m rsync -r dags/ gs://${{ secrets.COMPOSER_BUCKET }}/dags/

• Composer: gsutil rsync.
• MWAA: aws s3 sync.
• Astronomer: astro deploy.
• Kubernetes: Git plus git-sync.

Let Airflow retrieve DAGs from Git

[dag_processor]
dag_bundle_config_list = [
  {
    "name": "my_git_bundle",
    "classpath": "airflow.providers.git.bundles.git.GitDagBundle",
    "kwargs": {
      "tracking_ref": "main",
      "git_conn_id": "my_git_conn"
    }
  }
]

Verify the graph, not only the import

def test_dag_structure():
    dag = DagBag().get_dag("order_report")
    assert dag is not None
    assert len(dag.tasks) == 4

    extract = dag.get_task("extract_orders")
    downstream = {task.task_id for task in extract.downstream_list}
    assert "clean_orders" in downstream

Extract business logic from Airflow

def clean_orders_logic(orders: list) -> list:
    cleaned = [order for order in orders if order > 0]
    if not cleaned:
        raise ValueError("No valid orders remain")
    return cleaned

@task
def clean_orders(orders: list):
    return clean_orders_logic(orders)

Valid Python can still fail at runtime

BashOperator(
    task_id="print_date",
    bash_command="echo 'Processing {{ dss }}'",  # Typo.
)

• Import tests pass because the file is valid Python.
• Template rendering fails because dss is undefined.
• Rendered Template exposes the invalid expression.

Organize one DAG without creating a sub-DAG

with TaskGroup(group_id="transform_group") as transform_group:
    cleaned = clean(extract())
    enrich(cleaned)

transform_group >> load()

• TaskGroup changes visual organization.
• It does not create another DagRun.
• It replaces the removed SubDagOperator pattern.

Protect a shared resource from concurrency

airflow pools set postgres_pool 5 \
  "At most five production Postgres connections"

@task(pool="postgres_pool")
def query_postgres():
    hook = PostgresHook(postgres_conn_id="my_postgres")
    return hook.get_first("SELECT COUNT(*) FROM my_table")

Retries must not duplicate data

-- Anti-pattern: every retry inserts another row.
INSERT INTO daily_totals (report_date, total) VALUES (%s, %s);

-- Idempotent: one result per logical date.
INSERT INTO daily_totals (report_date, total) VALUES (%s, %s)
ON CONFLICT (report_date)
DO UPDATE SET total = EXCLUDED.total;

Use logical time for logical work

# Changes on every physical retry.
report["processed_at"] = datetime.now().isoformat()

# Stable for this logical run and its retries.
report["processed_at"] = context["ds"]

• Retries happen at a different wall-clock time.
• Backfills run historical intervals today.
• Logical date keeps output tied to the data interval.

Send context with every failure

def notify_on_failure(context):
    ti = context["task_instance"]
    print(
        f"Task {ti.dag_id}.{ti.task_id} failed. "
        f"Log: {ti.log_url}"
    )

@dag(default_args={"on_failure_callback": notify_on_failure}, ...)
def monitored_pipeline():
    ...

Apply least privilege per DAG

@dag(
    schedule="@daily",
    access_control={
        "data_team": {"can_read", "can_edit", "can_delete"},
        "analytics_viewers": {"can_read"},
    },
)
def restricted_dag():
    ...

Externalize secrets without changing DAG code

[secrets]
backend = airflow.providers.hashicorp.secrets.vault.VaultBackend
backend_kwargs = {
  "connections_path": "airflow/connections",
  "variables_path": "airflow/variables",
  "url": "https://vault.example.com:8200"
}

• The stable conn_id remains unchanged.
• Vault replaces metadata-DB lookup.
• AWS, GCP, and Azure provide equivalent backends.