/dag-factory

name: dag-factory
description: Authors Apache Airflow DAGs declaratively from dag-factory YAML configs. Use when building DAGs declaratively from YAML via dag-factory; creating/editing dag-factory templates/YAML configs,reating/editing dag-factory YAML configs, defaults, dynamic tasks, datasets, or callbacks; or validating dag-factory configurations; upgrading or re-pinning dag-factory.

DAG Factory

You are helping a user build Apache Airflow DAGs declaratively with **dag-factory**, a library that turns YAML configuration files into Airflow DAGs. Execute steps in order and prefer the simplest configuration that meets the user's needs.

> **Package**: `dag-factory` on PyPI > **Repo**: https://github.com/astronomer/dag-factory > **Docs**: https://astronomer.github.io/dag-factory/latest/ > **Targets**: dag-factory **v1.0+** only. For pre-1.0 projects, see [reference/migration.md](reference/migration.md) before applying any guidance from this skill. > **Requires**: Python 3.10+, Airflow 2.4+ (Airflow 3 supported)

Before Starting

Confirm with the user: 1. **Airflow version** ≥2.4 2. **Python version** ≥3.10 3. **dag-factory version**: this skill targets **v1.0+**. If the project is on <1.0, follow [reference/migration.md](reference/migration.md) to upgrade before continuing. 4. **Use case**: dag-factory is for declarative, low-code DAG authoring. If the user needs reusable, validated Pythonic templates with Pydantic, suggest **blueprint** instead. If they need full Python flexibility, suggest the **authoring-dags** skill.

---

Determine What the User Needs

| User Request | Action | |--------------|--------| | "Create a YAML DAG" / "Convert this Python DAG to YAML" | Go to **Defining a DAG in YAML** | | "Set up dag-factory in my project" | Go to **Project Setup** | | "Share defaults across DAGs" / "Set start_date once" | Go to **Defaults** | | "Use a custom operator" / "Use KPO / Slack / Snowflake" | Go to **Custom & Provider Operators** | | "Dynamic / mapped tasks" / "expand / partial" | Go to **Dynamic Task Mapping** | | "Schedule on dataset" / "Outlets and inlets" | Go to **Datasets** | | "Add a callback" / "Slack on failure" | Go to **Callbacks** | | "Use a timetable" / "datetime in YAML" / "timedelta in YAML" | Go to **Custom Python Objects (`__type__`)** | | "Lint my YAML" / "Validate" | Go to **Validation Commands** | | "Convert Airflow 2 YAML to Airflow 3" | Go to **Validation Commands** (`dagfactory convert`) | | "Migrate from dag-factory <1.0" | See [reference/migration.md](reference/migration.md) | | dag-factory errors / troubleshooting | Go to **Troubleshooting** |

---

Project Setup

1. Install the Package

Add to `requirements.txt`:

dag-factory>=1.0.0

dag-factory **does not** install Airflow providers automatically. Install any provider packages your YAML references (e.g., `apache-airflow-providers-slack`, `apache-airflow-providers-cncf-kubernetes`).

2. Create the Loader

Create `dags/load_dags.py` so Airflow's DAG processor will pick it up:

import os
from pathlib import Path

from dagfactory import load_yaml_dags

CONFIG_ROOT_DIR = Path(os.getenv("CONFIG_ROOT_DIR", "/usr/local/airflow/dags/"))

# Option A: load every *.yml / *.yaml under a folder
load_yaml_dags(globals_dict=globals(), dags_folder=str(CONFIG_ROOT_DIR))

# Option B: load a single file
# load_yaml_dags(globals_dict=globals(), config_filepath=str(CONFIG_ROOT_DIR / "my_dag.yml"))

# Option C: load from an in-Python dict
# load_yaml_dags(globals_dict=globals(), config_dict={...})

`globals_dict=globals()` is required so generated DAG objects are registered into the module namespace where Airflow can discover them.

3. Verify Installation

dagfactory --version

---

Defining a DAG in YAML

Each top-level YAML key (other than `default`) defines a DAG. The key becomes the `dag_id`. **Use the list format for `tasks` and `task_groups`** — it is the recommended format since v1.0.0.

# dags/example_dag_factory.yml
default:
  default_args:
    start_date: 2024-11-11

basic_example_dag:
  default_args:
    owner: "custom_owner"
  description: "this is an example dag"
  schedule: "0 3 * * *"
  catchup: false
  task_groups:
    - group_name: "example_task_group"
      tooltip: "this is an example task group"
      dependencies: [task_1]
  tasks:
    - task_id: "task_1"
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo 1"
    - task_id: "task_2"
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo 2"
      dependencies: [task_1]
    - task_id: "task_3"
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo 3"
      dependencies: [task_1]
      task_group_name: "example_task_group"

Key Fields

| Field | Where | Purpose | |-------|-------|---------| | `default` | top-level | Shared DAG-level args applied to every DAG in this file | | `default_args` | DAG or `default` block | Standard Airflow `default_args` (owner, retries, start_date, ...) | | `schedule` | DAG | Cron expression, preset (`@daily`), Dataset list, or `__type__` timetable | | `catchup` / `description` / `tags` | DAG | Standard Airflow DAG kwargs | | `tasks` | DAG | List of task dicts; each requires `task_id` and `operator` | | `operator` | task | **Full import path** to operator class (e.g. `airflow.operators.bash.BashOperator`) | | `dependencies` | task / task_group | List of upstream `task_id`s or `group_name`s | | `task_groups` | DAG | List of group dicts; each requires `group_name` | | `task_group_name` | task | Assigns a task to a task group |

Tasks do **not** need to be ordered by dependency in the YAML — dag-factory resolves the DAG topology.

Dictionary Format (Legacy)

Pre-1.0 dictionary format (where `tasks` is a dict keyed by `task_id`) still works for backward compatibility, but prefer the list format for new code.

---

Defaults

There are four ways to set defaults, in **precedence