Dynamic DAG Generation: Building Programmatic, Metadata-Driven Pipeline Factories at Scale

Introduction

Picture this: your data team has onboarded fifteen new source systems in the past quarter. Each one needs ingestion, validation, transformation, and load pipelines — all following roughly the same pattern but with enough differences in schema, schedule, SLA, and destination to make copy-paste development a genuine nightmare. Your Airflow project now has 47 DAG files, most of which are structural duplicates with different variable names. When a senior engineer changes the retry logic on one DAG, somebody has to manually propagate that change across every other file. A bug fix becomes a 40-file PR that nobody wants to review. Configuration drift sets in. Pipelines diverge. Your on-call rotation starts to feel like archaeology.

This is the problem dynamic DAG generation solves — and it solves it elegantly. Instead of writing individual DAG files per pipeline, you build a DAG factory: a system where metadata (stored in YAML, a database, or an API response) drives the programmatic creation of Airflow DAGs at parse time. Add a new row to a config table, and a new pipeline appears in your Airflow UI. Change a retry policy in one place, and every pipeline inherits it immediately. The factory pattern separates what a pipeline does from how individual instances of that pipeline are configured.

By the end of this lesson, you'll be able to design and implement production-grade dynamic DAG generation systems in Apache Airflow. We'll cover everything from the fundamentals of Python-level DAG factories to metadata-driven generation from external sources, with real attention to the performance implications, failure modes, and architectural trade-offs that most tutorials gloss over entirely.

What you'll learn:

• How Airflow's DAG discovery mechanism works, and how to exploit it for programmatic generation
• How to build type-safe, validated configuration schemas for your pipeline metadata
• How to generate entire DAG families from YAML files, database tables, and REST APIs
• How to handle performance, caching, and security concerns in dynamic generation at scale
• How to test, version, and operationalize a DAG factory in a real team environment

---

Prerequisites

You should be comfortable with:

• Apache Airflow fundamentals (DAGs, Operators, TaskFlow API, Connections, Variables)
• Python intermediate concepts: dataclasses, type hints, decorators, context managers
• Basic SQL and YAML
• A working Airflow 2.x environment (2.6+ preferred; examples use 2.8 features where noted)

You don't need to have built DAG factories before — that's what this lesson is for.

---

How Airflow Discovers and Parses DAGs

Before you can build a factory, you need to understand exactly what you're exploiting. Airflow's scheduler runs a process called the DagFileProcessor, which periodically scans your dags/ directory for Python files. For each file, it executes the file and looks for DAG objects in the module's global namespace. That's the critical insight: Airflow doesn't care how those DAG objects got there. They can be hand-authored, generated by a loop, returned from a factory function — anything. As long as a DAG instance exists at the module level when the file finishes executing, Airflow will pick it up.

The scheduler re-parses DAG files on a configurable interval (min_file_process_interval, defaulting to 30 seconds). This means your factory code runs repeatedly and frequently. Every time the scheduler re-parses your factory file, it's executing your Python code from scratch. This has enormous implications for performance that we'll dig into heavily later.

# Airflow scans for any variable in module globals that is a DAG instance
# These are equivalent from Airflow's perspective:

# Approach 1: Hand-authored
from airflow import DAG
dag = DAG("my_dag", ...)

# Approach 2: Factory function puts DAG in global namespace
def create_dag(dag_id):
    with DAG(dag_id, ...) as dag:
        # define tasks
        pass
    return dag

generated_dag = create_dag("my_dynamic_dag")

# Approach 3: Loop over config, inject multiple DAGs into globals
for config in pipeline_configs:
    globals()[config["dag_id"]] = create_dag(config)

That globals() injection pattern is the workhorse of DAG factories. Airflow scans the module's __dict__, and globals() gives you direct write access to it. Every DAG you inject there becomes visible to the scheduler.

Important: Airflow also looks for DAGs inside Python files via the .airflowignore mechanism and file-level dag_discovery_safe_mode. If safe_mode is True (default), Airflow will only parse files containing the string "airflow" and "DAG" somewhere in their text before executing them. Your factory file will naturally contain both, so this is rarely an issue — but be aware of it if you're using exotic file structures.

---

The Anatomy of a DAG Factory

Let's start with the simplest viable factory before layering in sophistication. Suppose you're building ingestion pipelines for a set of operational databases — let's say twelve regional e-commerce databases that all have the same schema but different connection IDs and load schedules.

# dags/regional_ingestion_factory.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.amazon.aws.transfers.sql_to_s3 import SqlToS3Operator

PIPELINE_CONFIGS = [
    {
        "dag_id": "ingest_ecommerce_us_east",
        "region": "us-east-1",
        "conn_id": "postgres_us_east",
        "s3_bucket": "datalake-us-east",
        "schedule": "0 2 * * *",
        "tables": ["orders", "customers", "products", "inventory"],
        "sla_minutes": 60,
    },
    {
        "dag_id": "ingest_ecommerce_eu_west",
        "region": "eu-west-1",
        "conn_id": "postgres_eu_west",
        "s3_bucket": "datalake-eu-west",
        "schedule": "0 3 * * *",
        "tables": ["orders", "customers", "products"],
        "sla_minutes": 90,
    },
    # ... 10 more regions
]


def create_ingestion_dag(config: dict) -> DAG:
    default_args = {
        "owner": "data-engineering",
        "retries": 3,
        "retry_delay": timedelta(minutes=5),
        "retry_exponential_backoff": True,
        "sla": timedelta(minutes=config["sla_minutes"]),
        "email_on_failure": True,
        "email": ["data-alerts@company.com"],
    }

    with DAG(
        dag_id=config["dag_id"],
        default_args=default_args,
        start_date=datetime(2024, 1, 1),
        schedule_interval=config["schedule"],
        catchup=False,
        tags=["ingestion", "regional", config["region"]],
        max_active_runs=1,
    ) as dag:

        # Dynamically create tasks for each table in this region's config
        previous_task = None
        for table in config["tables"]:
            extract_task = SqlToS3Operator(
                task_id=f"extract_{table}",
                sql=f"SELECT * FROM {table} WHERE updated_at > '{{{{ ds }}}}'",
                s3_bucket=config["s3_bucket"],
                s3_key=f"raw/{config['region']}/{table}/{{{{ ds }}}}/data.parquet",
                sql_conn_id=config["conn_id"],
                replace=True,
            )

            if previous_task:
                previous_task >> extract_task
            previous_task = extract_task

    return dag


# Inject all generated DAGs into module globals
for pipeline_config in PIPELINE_CONFIGS:
    dag_instance = create_ingestion_dag(pipeline_config)
    globals()[pipeline_config["dag_id"]] = dag_instance

This works, but it has several problems we need to fix before calling it production-ready. The config lives in the Python file itself (brittle), there's no validation (a typo in conn_id fails silently until runtime), and there's no parallelism within the extraction phase (one table must finish before the next starts).

Let's address each of these progressively.

---

Separating Configuration from Code with YAML

The next step toward a real factory is externalizing your metadata. YAML is the natural choice for configuration that data engineers (or even analysts) might need to edit — it's readable, supports comments, and has mature Python parsing support.

Create a directory structure like this:

dags/
  factory/
    __init__.py
    dag_builder.py
    schema.py
  pipeline_configs/
    ecommerce_us_east.yaml
    ecommerce_eu_west.yaml
    ecommerce_apac.yaml
  regional_ingestion_factory.py

Here's a well-structured YAML config file:

# pipeline_configs/ecommerce_us_east.yaml
dag_id: ingest_ecommerce_us_east
owner: data-engineering
region: us-east-1
conn_id: postgres_us_east
destination:
  type: s3
  bucket: datalake-us-east
  prefix: raw/us-east-1

schedule: "0 2 * * *"
sla_minutes: 60
max_active_runs: 1
catchup: false

tables:
  - name: orders
    primary_key: order_id
    incremental_column: updated_at
    parallelism: high
  - name: customers
    primary_key: customer_id
    incremental_column: updated_at
    parallelism: medium
  - name: products
    primary_key: product_id
    incremental_column: modified_at
    parallelism: low
  - name: inventory
    primary_key: sku_id
    incremental_column: last_checked
    parallelism: high

alerting:
  on_failure: true
  on_sla_miss: true
  channels:
    - email: data-alerts@company.com
    - slack: "#data-alerts"

Now write a typed schema using Python dataclasses (or Pydantic if you're in that ecosystem — we'll use dataclasses here to minimize dependencies):

# dags/factory/schema.py
from dataclasses import dataclass, field
from typing import List, Literal, Optional
from enum import Enum


class Parallelism(str, Enum):
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"


PARALLELISM_POOL_SLOTS = {
    Parallelism.HIGH: 4,
    Parallelism.MEDIUM: 2,
    Parallelism.LOW: 1,
}


@dataclass
class TableConfig:
    name: str
    primary_key: str
    incremental_column: str
    parallelism: Parallelism = Parallelism.MEDIUM

    def __post_init__(self):
        self.parallelism = Parallelism(self.parallelism)


@dataclass
class DestinationConfig:
    type: Literal["s3", "gcs", "adls"]
    bucket: str
    prefix: str


@dataclass
class AlertingConfig:
    on_failure: bool = True
    on_sla_miss: bool = True
    channels: List[dict] = field(default_factory=list)

    def get_emails(self) -> List[str]:
        return [
            ch["email"] for ch in self.channels if "email" in ch
        ]


@dataclass
class PipelineConfig:
    dag_id: str
    owner: str
    region: str
    conn_id: str
    destination: DestinationConfig
    tables: List[TableConfig]
    schedule: str
    sla_minutes: int
    max_active_runs: int = 1
    catchup: bool = False
    alerting: AlertingConfig = field(default_factory=AlertingConfig)

    def __post_init__(self):
        # Validate that dag_id follows our naming convention
        if not self.dag_id.startswith("ingest_"):
            raise ValueError(
                f"dag_id '{self.dag_id}' must start with 'ingest_' "
                f"to ensure proper categorization in the UI"
            )
        if self.sla_minutes < 15:
            raise ValueError(
                f"SLA of {self.sla_minutes} minutes is unrealistically short. "
                f"Minimum is 15 minutes."
            )

    @classmethod
    def from_dict(cls, data: dict) -> "PipelineConfig":
        """Deserialize a raw dictionary (from YAML) into a validated PipelineConfig."""
        destination = DestinationConfig(**data.pop("destination"))
        tables = [TableConfig(**t) for t in data.pop("tables")]
        alerting_data = data.pop("alerting", {})
        alerting = AlertingConfig(**alerting_data)

        return cls(
            destination=destination,
            tables=tables,
            alerting=alerting,
            **data
        )

Now write the loader that reads YAML files and produces validated configs:

# dags/factory/__init__.py
import os
import yaml
from pathlib import Path
from typing import List
from .schema import PipelineConfig

CONFIG_DIR = Path(__file__).parent.parent / "pipeline_configs"


def load_pipeline_configs(config_dir: Path = CONFIG_DIR) -> List[PipelineConfig]:
    """
    Scan the config directory for YAML files and return validated PipelineConfig objects.
    Failures in individual config files are logged but don't block other configs from loading.
    """
    configs = []
    errors = []

    for yaml_file in sorted(config_dir.glob("*.yaml")):
        try:
            with open(yaml_file) as f:
                raw = yaml.safe_load(f)
            config = PipelineConfig.from_dict(raw)
            configs.append(config)
        except Exception as e:
            errors.append(f"{yaml_file.name}: {e}")

    if errors:
        # Log but don't raise — we want valid configs to still load
        import logging
        logger = logging.getLogger(__name__)
        for error in errors:
            logger.error("Failed to load pipeline config: %s", error)

    return configs

Design decision: We catch exceptions per-file rather than letting one bad config file break the entire factory. This is a deliberate choice — if you have 20 pipeline configs and one has a typo, you want the other 19 to still work. Log the failure loudly, alert on it, but don't cascade it.

---

Building the DAG Builder Module

Now we have typed, validated configs. Let's build the actual DAG construction logic as a clean, testable module:

# dags/factory/dag_builder.py
from datetime import datetime, timedelta
from typing import Dict
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.transfers.sql_to_s3 import SqlToS3Operator
from airflow.utils.task_group import TaskGroup
from .schema import PipelineConfig, Parallelism, PARALLELISM_POOL_SLOTS


def _build_default_args(config: PipelineConfig) -> dict:
    return {
        "owner": config.owner,
        "retries": 3,
        "retry_delay": timedelta(minutes=5),
        "retry_exponential_backoff": True,
        "max_retry_delay": timedelta(minutes=60),
        "sla": timedelta(minutes=config.sla_minutes),
        "email_on_failure": config.alerting.on_failure,
        "email": config.alerting.get_emails(),
    }


def _get_pool_for_parallelism(parallelism: Parallelism) -> str:
    """
    Map parallelism levels to Airflow pool names.
    Pools must be pre-created in the Airflow UI or via the API.
    """
    pool_map = {
        Parallelism.HIGH: "ingestion_high_priority",
        Parallelism.MEDIUM: "ingestion_medium_priority",
        Parallelism.LOW: "ingestion_low_priority",
    }
    return pool_map[parallelism]


def build_ingestion_dag(config: PipelineConfig) -> DAG:
    """
    Construct an Airflow DAG from a PipelineConfig.

    This function is the heart of the factory. It's kept pure (no side effects
    beyond DAG object creation) so it can be unit tested without an Airflow environment.
    """
    default_args = _build_default_args(config)

    with DAG(
        dag_id=config.dag_id,
        default_args=default_args,
        start_date=datetime(2024, 1, 1),
        schedule_interval=config.schedule,
        catchup=config.catchup,
        tags=["ingestion", "regional", config.region, "factory-generated"],
        max_active_runs=config.max_active_runs,
        doc_md=f"""
        ## {config.dag_id}

        **Auto-generated** by the regional ingestion factory.

        | Property | Value |
        |----------|-------|
        | Region | `{config.region}` |
        | Source Connection | `{config.conn_id}` |
        | Destination | `{config.destination.type}://{config.destination.bucket}/{config.destination.prefix}` |
        | SLA | {config.sla_minutes} minutes |
        | Tables | {len(config.tables)} |

        *Modify `pipeline_configs/{config.dag_id.replace('ingest_', '')}.yaml` to change this pipeline.*
        """,
    ) as dag:

        with TaskGroup("extract_tables") as extract_group:
            for table_config in config.tables:
                SqlToS3Operator(
                    task_id=f"extract_{table_config.name}",
                    sql=(
                        f"SELECT * FROM {table_config.name} "
                        f"WHERE {table_config.incremental_column} >= '{{{{ data_interval_start }}}}' "
                        f"AND {table_config.incremental_column} < '{{{{ data_interval_end }}}}'"
                    ),
                    s3_bucket=config.destination.bucket,
                    s3_key=(
                        f"{config.destination.prefix}/"
                        f"{table_config.name}/"
                        f"{{{{ ds }}}}/data.parquet"
                    ),
                    sql_conn_id=config.conn_id,
                    replace=True,
                    pool=_get_pool_for_parallelism(table_config.parallelism),
                    doc_md=f"Extract `{table_config.name}` incrementally on `{table_config.incremental_column}`",
                )
            # Note: Tasks in a TaskGroup without explicit dependencies run in parallel by default

        def validate_extracts(**context):
            """
            Check that all expected S3 objects were created.
            In practice, you'd use S3Hook to verify object existence and size.
            """
            ds = context["ds"]
            expected_paths = [
                f"{config.destination.prefix}/{t.name}/{ds}/data.parquet"
                for t in config.tables
            ]
            # Simulate validation - replace with real S3Hook checks
            import logging
            logging.getLogger(__name__).info(
                "Validating %d extracted files for %s",
                len(expected_paths),
                config.dag_id
            )

        validate = PythonOperator(
            task_id="validate_extracts",
            python_callable=validate_extracts,
        )

        extract_group >> validate

    return dag

Notice a few intentional design choices here:

• Tasks within the extract_tables TaskGroup run in parallel by default — no explicit dependencies between them. This is a major improvement over the sequential approach in our naive first example.
• Each task gets assigned to an Airflow pool based on its declared parallelism level, giving you fine-grained concurrency control across all generated DAGs.
• The DAG's doc_md is auto-generated from the config, giving operators useful documentation without any manual effort.

---

The Factory Entry Point

Now wire everything together in the factory file:

# dags/regional_ingestion_factory.py
"""
Regional Ingestion DAG Factory

This file generates one Airflow DAG per YAML config in pipeline_configs/.
To add a new pipeline, create a new YAML file — no Python changes required.
"""
import logging
from factory import load_pipeline_configs
from factory.dag_builder import build_ingestion_dag

logger = logging.getLogger(__name__)

pipeline_configs = load_pipeline_configs()
logger.info("Loaded %d pipeline configurations", len(pipeline_configs))

for config in pipeline_configs:
    try:
        dag = build_ingestion_dag(config)
        globals()[config.dag_id] = dag
        logger.debug("Registered DAG: %s", config.dag_id)
    except Exception as e:
        logger.error(
            "Failed to build DAG for config '%s': %s",
            config.dag_id,
            e,
            exc_info=True
        )

This is now a proper factory. Add a YAML file → new DAG appears. Edit YAML → pipeline changes on next parse cycle. Remove YAML → DAG disappears (and Airflow marks it as removed in the DB).

---

Driving the Factory from a Database

YAML files are great for team-managed configs, but sometimes your pipeline metadata lives in a database — a governance catalog, a data product registry, or an operational config table. Driving generation from a database lets non-engineers (via an internal UI, for example) register new pipelines without touching files.

The key challenge is performance. Querying a database on every DAG parse cycle (every 30 seconds) would hammer your config DB, especially as you scale. The solution is aggressive caching with TTL-based invalidation.

# dags/factory/db_loader.py
import logging
import time
from functools import lru_cache
from typing import List, Optional
import json

logger = logging.getLogger(__name__)

# We cache the database response for 5 minutes to avoid hammering the config DB
# on every Airflow scheduler parse cycle
_CACHE_TTL_SECONDS = 300
_cache: dict = {"data": None, "fetched_at": 0.0}


def _is_cache_valid() -> bool:
    return (
        _cache["data"] is not None
        and (time.monotonic() - _cache["fetched_at"]) < _CACHE_TTL_SECONDS
    )


def load_configs_from_database(conn_id: str = "postgres_config_db") -> List[dict]:
    """
    Load pipeline configurations from a relational config table.

    Uses a simple time-based cache to avoid re-querying on every scheduler
    parse cycle. The cache is process-local, so each Airflow worker/scheduler
    process maintains its own cache.
    """
    if _is_cache_valid():
        logger.debug("Returning cached pipeline configs (%d records)", len(_cache["data"]))
        return _cache["data"]

    try:
        # Use Airflow's hook system to respect Connection management
        from airflow.providers.postgres.hooks.postgres import PostgresHook
        hook = PostgresHook(postgres_conn_id=conn_id)

        records = hook.get_records(
            """
            SELECT
                dag_id,
                owner,
                region,
                conn_id,
                destination_config,
                table_configs,
                schedule,
                sla_minutes,
                max_active_runs,
                catchup,
                alerting_config,
                is_active
            FROM pipeline_registry.dag_configs
            WHERE is_active = true
            ORDER BY dag_id
            """
        )

        configs = []
        for row in records:
            (dag_id, owner, region, conn_id_val, dest_json,
             tables_json, schedule, sla_minutes, max_active_runs,
             catchup, alerting_json, is_active) = row

            config = {
                "dag_id": dag_id,
                "owner": owner,
                "region": region,
                "conn_id": conn_id_val,
                "destination": json.loads(dest_json),
                "tables": json.loads(tables_json),
                "schedule": schedule,
                "sla_minutes": sla_minutes,
                "max_active_runs": max_active_runs,
                "catchup": catchup,
                "alerting": json.loads(alerting_json) if alerting_json else {},
            }
            configs.append(config)

        _cache["data"] = configs
        _cache["fetched_at"] = time.monotonic()
        logger.info("Loaded %d active pipeline configs from database", len(configs))
        return configs

    except Exception as e:
        logger.error("Failed to load configs from database: %s", e, exc_info=True)
        # Return stale cache if available — fail open rather than disrupting all DAGs
        if _cache["data"] is not None:
            logger.warning("Returning stale cache due to database error")
            return _cache["data"]
        return []

Fail-open vs. fail-closed: This loader deliberately returns stale cached data if the database is unavailable. For pipeline configuration, failing open (returning old configs) is almost always better than failing closed (returning nothing and causing all DAGs to disappear). Tune this decision based on your operational risk profile.

The database table that backs this would look like:

CREATE TABLE pipeline_registry.dag_configs (
    dag_id          VARCHAR(250) PRIMARY KEY,
    owner           VARCHAR(100) NOT NULL,
    region          VARCHAR(50)  NOT NULL,
    conn_id         VARCHAR(100) NOT NULL,
    destination_config  JSONB    NOT NULL,
    table_configs   JSONB        NOT NULL,
    schedule        VARCHAR(100) NOT NULL,
    sla_minutes     INT          NOT NULL DEFAULT 60,
    max_active_runs INT          NOT NULL DEFAULT 1,
    catchup         BOOLEAN      NOT NULL DEFAULT FALSE,
    alerting_config JSONB,
    is_active       BOOLEAN      NOT NULL DEFAULT TRUE,
    created_at      TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    created_by      VARCHAR(100),
    notes           TEXT
);

-- Index for fast active config lookups
CREATE INDEX idx_dag_configs_active ON pipeline_registry.dag_configs(is_active)
WHERE is_active = true;

Security note: The connection used to query your config database (postgres_config_db) should have read-only access to pipeline_registry.dag_configs only. The Airflow scheduler runs under a service account — it should never have write access to your config tables. Enforce this with database role grants, not just application-level logic.

---

Advanced Pattern: Template-Based DAG Families

As your factory matures, you'll encounter pipelines that share structure but have meaningfully different workflow logic — not just different parameters. An ingestion pipeline looks very different from a dbt transformation pipeline or a reverse ETL pipeline. The solution is a template registry that maps a pipeline_type field in your config to a specific builder function.

# dags/factory/template_registry.py
from typing import Callable, Dict
from airflow import DAG
from .schema import PipelineConfig

# Type alias for builder functions
DagBuilder = Callable[[PipelineConfig], DAG]

_REGISTRY: Dict[str, DagBuilder] = {}


def register_template(pipeline_type: str):
    """
    Decorator that registers a DAG builder function for a given pipeline type.

    Usage:
        @register_template("ingestion")
        def build_ingestion_dag(config: PipelineConfig) -> DAG:
            ...
    """
    def decorator(func: DagBuilder) -> DagBuilder:
        if pipeline_type in _REGISTRY:
            raise ValueError(
                f"Template '{pipeline_type}' is already registered. "
                f"Each pipeline type must have exactly one builder."
            )
        _REGISTRY[pipeline_type] = func
        return func
    return decorator


def get_builder(pipeline_type: str) -> DagBuilder:
    if pipeline_type not in _REGISTRY:
        available = ", ".join(sorted(_REGISTRY.keys()))
        raise KeyError(
            f"No builder registered for pipeline type '{pipeline_type}'. "
            f"Available types: {available}"
        )
    return _REGISTRY[pipeline_type]

Now in each builder module, you decorate with @register_template:

# dags/factory/builders/ingestion.py
from ..template_registry import register_template
from ..schema import PipelineConfig
from airflow import DAG

@register_template("ingestion")
def build_ingestion_dag(config: PipelineConfig) -> DAG:
    # ... ingestion-specific DAG construction
    pass


# dags/factory/builders/dbt_transformation.py
from ..template_registry import register_template

@register_template("dbt_transformation")
def build_dbt_dag(config: PipelineConfig) -> DAG:
    # Different structure: uses DbtRunOperator, DbtTestOperator, etc.
    pass


# dags/factory/builders/reverse_etl.py
from ..template_registry import register_template

@register_template("reverse_etl")
def build_reverse_etl_dag(config: PipelineConfig) -> DAG:
    # Different again: reads from warehouse, writes to operational DB or SaaS
    pass

The factory entry point becomes type-agnostic:

# dags/pipeline_factory.py
from factory import load_pipeline_configs
from factory.template_registry import get_builder
# Import all builders so their @register_template decorators fire
import factory.builders.ingestion
import factory.builders.dbt_transformation
import factory.builders.reverse_etl

import logging
logger = logging.getLogger(__name__)

for config in load_pipeline_configs():
    try:
        builder = get_builder(config.pipeline_type)
        dag = builder(config)
        globals()[config.dag_id] = dag
    except KeyError as e:
        logger.error("Unknown pipeline type for %s: %s", config.dag_id, e)
    except Exception as e:
        logger.error("Failed building DAG %s: %s", config.dag_id, e, exc_info=True)

This pattern scales cleanly. Adding a new pipeline type is a matter of writing a new builder module and registering it — the factory file doesn't change.

---

Performance at Scale: The Parse-Time Cost Problem

Here's where expert-level thinking separates production factories from toy examples. Every DAG file in Airflow is parsed every min_file_process_interval seconds. If you have one factory file generating 50 DAGs, that file is parsed every 30 seconds. If parsing that file involves a database query that takes 200ms, that's one query every 30 seconds — manageable. If you have ten factory files, that's ten queries every 30 seconds. And if Airflow is running multiple DagFileProcessor workers...

Let's think through the math. With parsing_processes = 4 (the Airflow default) and min_file_process_interval = 30, each file gets parsed at most once per 30 seconds, but across all files the aggregate parse rate is much higher. At 50 DAG factory files, you might be looking at thousands of parse operations per hour.

Profiling your parse time:

# Add this to your factory file temporarily to measure parse cost
import time
import logging

_parse_start = time.perf_counter()

# ... your factory code ...

_parse_duration = time.perf_counter() - _parse_start
logging.getLogger(__name__).info(
    "DAG factory parse completed in %.3fs, generated %d DAGs",
    _parse_duration,
    len([k for k, v in globals().items() if hasattr(v, 'dag_id')])
)

Strategies for keeping parse time low:

1. Aggressive caching (as shown in the database loader above). The parse-time cache should outlive multiple parse cycles. Don't query external systems on every parse.

2. Keep the factory file count low. One factory file per category of pipeline (ingestion, transformation, etc.), not one per source system. The per-file parse overhead adds up.

3. Use Airflow Variables sparingly. Calling Variable.get() at DAG parse time triggers a database query. If you're generating 50 DAGs and each one calls Variable.get() three times, that's 150 Airflow database queries per parse cycle. Pre-fetch all variables at the top of the file, once.

# BAD: Variable fetched per DAG, inside the loop
for config in configs:
    with DAG(...) as dag:
        env = Variable.get("environment")  # This queries the DB 50 times per parse!
        ...

# GOOD: Fetch once, use everywhere
ENVIRONMENT = Variable.get("environment", default_var="production")
for config in configs:
    with DAG(...) as dag:
        # Use ENVIRONMENT directly
        ...

4. Avoid heavy computation at parse time. Parse time is for constructing the DAG object graph, not for doing the actual work of your pipeline. If you're tempted to call an external API, run SQL, or load a large file at parse time, move that logic into a task.

5. Consider .airflowignore for factory file organization. Put non-DAG Python modules (your schema, builder, loader code) in directories that Airflow won't try to parse as DAG files. The factory/ subdirectory with an __init__.py won't be scanned for DAGs by default if it doesn't contain the "DAG" string, but it's cleaner to explicitly .airflowignore it.

---

Testing Your DAG Factory

Testing a DAG factory requires a different mindset than testing individual DAGs. You're testing both the generation mechanism (does the factory produce the right number of DAGs with the right structure?) and the individual DAG logic (do tasks have correct parameters?).

# tests/test_dag_factory.py
import pytest
from pathlib import Path
from unittest.mock import patch, MagicMock
from factory.schema import PipelineConfig, TableConfig, DestinationConfig
from factory.dag_builder import build_ingestion_dag


@pytest.fixture
def sample_config():
    return PipelineConfig(
        dag_id="ingest_test_region",
        owner="test-team",
        region="us-test-1",
        conn_id="postgres_test",
        destination=DestinationConfig(
            type="s3",
            bucket="test-bucket",
            prefix="raw/us-test-1"
        ),
        tables=[
            TableConfig(
                name="orders",
                primary_key="order_id",
                incremental_column="updated_at",
                parallelism="high"
            ),
            TableConfig(
                name="customers",
                primary_key="customer_id",
                incremental_column="updated_at",
                parallelism="medium"
            ),
        ],
        schedule="0 2 * * *",
        sla_minutes=60,
    )


def test_dag_is_created(sample_config):
    """Factory should produce a valid DAG object."""
    dag = build_ingestion_dag(sample_config)
    assert dag is not None
    assert dag.dag_id == "ingest_test_region"


def test_dag_has_correct_schedule(sample_config):
    dag = build_ingestion_dag(sample_config)
    assert dag.schedule_interval == "0 2 * * *"


def test_dag_has_correct_task_count(sample_config):
    """Should have one extract task per table plus the validate task."""
    dag = build_ingestion_dag(sample_config)
    task_ids = dag.task_ids
    assert "extract_tables.extract_orders" in task_ids
    assert "extract_tables.extract_customers" in task_ids
    assert "validate_extracts" in task_ids


def test_dag_has_no_cycles(sample_config):
    """Airflow's built-in cycle detection should pass."""
    dag = build_ingestion_dag(sample_config)
    assert dag.test_cycle() is False


def test_config_validation_rejects_short_sla():
    """Schema validation should catch invalid configs before DAG construction."""
    with pytest.raises(ValueError, match="Minimum is 15 minutes"):
        PipelineConfig(
            dag_id="ingest_test",
            owner="team",
            region="us-east-1",
            conn_id="postgres_test",
            destination=DestinationConfig(type="s3", bucket="b", prefix="p"),
            tables=[],
            schedule="@daily",
            sla_minutes=5,  # Too short!
        )


def test_config_validation_rejects_wrong_dag_id_prefix():
    with pytest.raises(ValueError, match="must start with 'ingest_'"):
        PipelineConfig(
            dag_id="wrong_prefix_dag",
            # ... rest of valid config
        )


def test_factory_loads_yaml_configs(tmp_path):
    """Factory loader should correctly parse a well-formed YAML config."""
    yaml_content = """
dag_id: ingest_test_loader
owner: test-team
region: us-test-1
conn_id: postgres_test
destination:
  type: s3
  bucket: test-bucket
  prefix: raw/test
schedule: "@daily"
sla_minutes: 30
tables:
  - name: orders
    primary_key: order_id
    incremental_column: updated_at
    parallelism: high
"""
    config_file = tmp_path / "test_pipeline.yaml"
    config_file.write_text(yaml_content)

    from factory import load_pipeline_configs
    configs = load_pipeline_configs(config_dir=tmp_path)

    assert len(configs) == 1
    assert configs[0].dag_id == "ingest_test_loader"
    assert len(configs[0].tables) == 1


def test_factory_skips_invalid_yaml(tmp_path):
    """A bad YAML file should not block valid configs from loading."""
    (tmp_path / "bad.yaml").write_text("dag_id: this_is_missing_required_fields")
    (tmp_path / "good.yaml").write_text("""
dag_id: ingest_good_pipeline
owner: team
region: us-east-1
conn_id: postgres
destination:
  type: s3
  bucket: bucket
  prefix: prefix
schedule: "@daily"
sla_minutes: 30
tables:
  - name: orders
    primary_key: id
    incremental_column: updated_at
""")
    from factory import load_pipeline_configs
    configs = load_pipeline_configs(config_dir=tmp_path)
    assert len(configs) == 1
    assert configs[0].dag_id == "ingest_good_pipeline"

Run these with pytest tests/test_dag_factory.py -v. None of these tests require a running Airflow environment — they test the pure Python logic, which is exactly what you want for fast CI feedback.

---

Hands-On Exercise

Build a dynamic DAG factory for a fictional e-commerce analytics platform. Here's your scenario:

Context: You're a data engineer at RetailPulse, an analytics company. You manage data pipelines for 8 client accounts. Each client has their own source database (PostgreSQL), their own S3 landing zone, and different SLAs and table sets. You need a DAG factory that:

1. Reads client configurations from YAML files in a client_configs/ directory
2. Generates one ingestion DAG per client
3. Each DAG should extract the tables listed in the client config, with parallel extraction
4. Validates that required S3 objects were created after extraction
5. Has client-specific retry policies (high-value clients get more retries)
6. Tags each DAG with the client name, tier (gold/silver/bronze), and "factory-generated"

Your YAML schema should support:

client_id: "acme_corp"
client_name: "ACME Corporation"
tier: "gold"  # gold/silver/bronze
conn_id: "postgres_acme"
s3_bucket: "retailpulse-acme-landing"
schedule: "0 */4 * * *"
sla_minutes: 45
tables:
  - name: transactions
    incremental_column: created_at
  - name: customers
    incremental_column: updated_at
  - name: products
    incremental_column: modified_at

Retry policy by tier:

• Gold: 5 retries, 2-minute delay
• Silver: 3 retries, 5-minute delay
• Bronze: 2 retries, 10-minute delay

Deliverables:

• A PipelineConfig dataclass with validation (tier must be gold/silver/bronze, dag_id derived from client_id)
• A build_client_dag() builder function
• A factory entry point at dags/client_ingestion_factory.py
• At least 3 YAML config files
• Tests for schema validation, DAG structure, and the loader's resilience to bad configs

Stretch goal: Add a ClientConfigLoader that falls back to YAML files if a database connection is unavailable, using the try DB first, fall back to YAML pattern.

---

Common Mistakes & Troubleshooting

Mistake 1: Using the same task_id across dynamically generated DAGs

This is subtle and maddening. If your factory loop has a bug that generates the same task_id twice within a single DAG, Airflow will silently use the last one. Always use the config-driven values (table name, region, etc.) as part of task IDs.

# BAD - task_id collision risk
for table in config.tables:
    extract = SqlToS3Operator(task_id="extract")  # All tasks named "extract"!

# GOOD
for table in config.tables:
    extract = SqlToS3Operator(task_id=f"extract_{table.name}")

Mistake 2: Global state mutation between factory runs

Your factory module is re-executed periodically. If you have mutable global state that accumulates across runs, you'll get unexpected behavior.

# BAD - list grows on every parse
GENERATED_DAGS = []
for config in configs:
    dag = build_dag(config)
    GENERATED_DAGS.append(dag)  # This list keeps growing in memory!

# GOOD - just inject into globals, don't maintain secondary state
for config in configs:
    globals()[config.dag_id] = build_dag(config)

Mistake 3: Late-binding closures in task callables

This is a classic Python closure bug that hits hard in DAG factories. If you define a Python callable inside a loop that references the loop variable, all tasks will use the last value of the loop variable.

# BAD - all tasks will use the last value of 'table'
for table in config.tables:
    def extract_data(**context):
        print(f"Extracting {table}")  # 'table' is captured by reference!

    PythonOperator(task_id=f"extract_{table}", python_callable=extract_data)

# GOOD - use a default argument to capture the current value
for table in config.tables:
    def extract_data(table_name=table.name, **context):  # Bound at definition time
        print(f"Extracting {table_name}")

    PythonOperator(task_id=f"extract_{table.name}", python_callable=extract_data)

# ALSO GOOD - use functools.partial
from functools import partial

def extract_data(table_name, **context):
    print(f"Extracting {table_name}")

for table in config.tables:
    PythonOperator(
        task_id=f"extract_{table.name}",
        python_callable=partial(extract_data, table_name=table.name)
    )

Mistake 4: Querying Airflow's metadata DB during parse time without connection pooling

If your factory calls Variable.get() or Connection.get_connection_from_secrets() in a tight loop, each call opens a new database connection. At scale, this can exhaust your connection pool. Fetch once at module level, cache the result.

Mistake 5: Not handling the case where a previously-generated DAG disappears

If you remove a config file or set is_active = false in your database, the DAG disappears from the factory's output. Airflow will mark it as "Missing" in the UI. Existing DAG runs are preserved, but new ones won't trigger. This is usually the desired behavior, but make sure your team understands it. Document the "DAG offboarding" process explicitly.

Troubleshooting: DAGs not appearing after adding a config

1. Check the scheduler logs for parse errors: airflow scheduler -l or look in your logging backend
2. Confirm the YAML file has no syntax errors: python -c "import yaml; yaml.safe_load(open('pipeline_configs/new.yaml'))"
3. Check that the factory file itself doesn't have a module-level exception that aborts the entire parse
4. Verify your config passes schema validation by running the loader directly in a Python shell
5. Check .airflowignore — ensure your config directory isn't being ignored

Troubleshooting: Intermittent "DAG not found" errors in task execution

This happens when the scheduler and workers have slightly different views of which DAGs exist — usually because worker processes have stale cached configs. If using the database loader pattern, verify your cache TTL is appropriate and that worker processes can reach the config database.

---

Summary & Next Steps

Dynamic DAG generation is one of those techniques that feels like an advanced trick until you've used it in production — then it feels indispensable. Let's recap what we built:

• A schema layer (Python dataclasses with validation) that makes invalid configurations fail fast and loudly at load time, not silently at runtime
• A YAML-based configuration system that separates pipeline metadata from pipeline logic
• A database-backed loader with TTL caching that prevents parse-time database hammering
• A template registry using the decorator pattern that makes the factory extensible without modification
• A test suite that validates the factory's behavior in isolation, without needing a live Airflow environment
• A clear understanding of the performance implications of parse-time code execution and how to work within them

The factory pattern fundamentally changes how you think about pipeline development. Instead of asking "how do I write this pipeline?" you ask "what's the right abstraction for this class of pipelines?" That shift in thinking pays dividends at scale.

Where to go next:

• Airflow Datasets and Data-Aware Scheduling: Dynamic factories pair naturally with Airflow 2.4+ Datasets — generate pipelines that trigger on data arrival rather than schedules, and express cross-DAG dependencies declaratively
• DAG versioning: Explore tools like astronomer-dag-versioning or implement your own version tracking in the config schema to understand which factory version generated a given DAG
• GitOps for pipeline configs: Treat your YAML config directory as a GitOps repo — PRs for new pipelines, automated schema validation in CI, automatic deployment on merge
• Airflow's REST API for dynamic registration: Instead of file-based DAG discovery, explore the newer patterns using Airflow's REST API to register pipelines programmatically
• Observability at scale: With 50+ generated DAGs, you need DAG-level SLI/SLO tracking, not just task-level logs. Look into integrating with Datadog, Grafana, or Great Expectations at the factory level

The measure of a great DAG factory isn't the elegance of the code — it's whether a non-engineer can add a new pipeline by editing a YAML file, and whether you can onboard a new source system in under an hour. Build toward that standard.