dag-factory

What this skill does

# 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:

```python
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

```bash
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.

```yaml
# 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 order** (highest first):

1. `default_args` / DAG-level keys inside an individual DAG
2. The top-level `default:` block in the same YAML file
3. `defaults_config_dict=` argument to `load_yaml_dags`
4. A `defaults.yml` (or `defaults.yaml`) file via `defaults_config_path=` (or auto-detected next to the DAG YAML)

> Note: loader argument names and several other field names changed in v1.0.0. See [reference/migration.md](reference/migration.md) if you're working on an older project.

### `default` Block in the Same File

Powerful for templating multiple DAGs from one file:

```yaml
default:
  default_args:
    owner: "data-team"
    start_date: 2025-01-01
    retries: 2
  catchup: false
  schedule: "@daily"

dag_one:
  description: "first DAG"
  tasks:
    - task_id: t1
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo one"

dag_two:
  description: "second DAG"
  tasks:
    - task_id: t1
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo two"
```

### `defaults.yml` File

Place a `defaults.yml` next to the DAG YAML, or point `defaults_config_path` at a parent directory. dag-factory **merges** all `defaults.yml` files walking up the directory tree, with the file closest to the DAG YAML winning. DAG-level args (e.g. `schedule`, `catchup`) go at the root of `defaults.yml`; per-task defaults go under `default_args`.

```yaml
# defaults.yml
schedule: 0 1 * * *
catchup: false
default_args:
  start_date: '2024-12-31'
  owner: data-team
```

---

## Custom & Provider Operators

Reference any operator by its **full Python import path**. dag-factory passes all other task keys as kwargs to that operator.

```yaml
tasks:
  - task_id: begin
    operator: airflow.providers.standard.operators.empty.EmptyOperator
  - task_id: make_bread
    operator: customized.operators.breakfast_operators.MakeBreadOperator
    bread_type: 'Sourdough'
```

The operator's package must be installed and importable. For Airflow 3, prefer `airflow.providers.standard.operators.*` over the legacy `airflow.operators.*` paths — the `dagfactory convert` CLI rewrites these automatically.

### KubernetesPodOperator

Specify the operator path and pass kwargs directly. As of v1.0, dag-factory no longer does legacy type casting — use `__type__` for nested k8s objects.

```yaml
tasks:
  - task_id: hello-world-pod
    operator: airflow.providers.cncf.k