Dag Factory

dag-factory

Declarative Airflow DAGs from YAML. Detect the repo's dialect before writing YAML. Official Astronomer skill targets PyPI dag-factory v1+ (list-format tasks). Many production repos still use a vendored plugin and map-format tasks keyed by task_id. Mixing the two breaks parse.

1. Detect dialect

rg -n "load_yaml_dags|from dagfactory|from plugins.dagfactory" --glob '*.py' dags plugins | head
rg -n "^  tasks:" -g '*.yaml' -g '*.yml' -A 6 | head -40

Match neighboring YAML in the same folder. Do not "modernize" map-format files to list-format unless the user asks and the loader is PyPI v1+.

Keep from airflow import DAG in every loader module even if it looks unused. The DAG processor requires it.

2. Loader

Map-style in-repo plugin (typical):

from airflow import DAG
from plugins.dagfactory import load_yaml_dags

load_yaml_dags(globals_dict=globals(), dags_folder=".../configs")

PyPI v1:

from airflow import DAG
from dagfactory import load_yaml_dags

load_yaml_dags(globals_dict=globals(), dags_folder="/usr/local/airflow/dags")

globals_dict=globals() is required.

3. YAML shape

Map format (in-repo / pre-1.0)

default:
  default_args:
    owner: data-team
    retries: 3
    retry_delay_sec: 180
    on_failure_callback: package.callbacks.task_failed
  on_failure_callback: package.callbacks.dag_failed
  catchup: false

elt_source_object:
  schedule: 0 2 * * *
  max_active_runs: 1
  params:
    stg_database: AIRFLOW_VARIABLE__source/stg_database
  tasks:
    extract:
      operator: package.operators.extract.ExtractOperator
      http_conn: source_conn
    load:
      operator: package.operators.load.LoadOperator
      dependencies:
        - extract

Task id = YAML key. dependencies is a list of upstream task ids (not >>).

List format (PyPI v1+)

default:
  default_args:
    start_date: 2025-01-01

elt_source_object:
  schedule: 0 2 * * *
  catchup: false
  tasks:
    - task_id: extract
      operator: airflow.providers.standard.operators.empty.EmptyOperator
    - task_id: load
      operator: airflow.providers.standard.operators.empty.EmptyOperator
      dependencies: [extract]

4. Hard rules

• Full operator import path. airflow.providers.standard.operators.python.PythonOperator, not a short name. Airflow 3: prefer airflow.providers.* over airflow.operators.*.
• Callbacks are string paths that the factory imports. Two layers: task default_args.on_failure_callback (after retries) plus DAG on_failure_callback (run summary). DAG-level does not fire on each failed task.
• Do not add on_retry_callback unless the user asks. Retries are expected.
• Params may use an AIRFLOW_VARIABLE__ prefix when the factory resolves Airflow Variables at parse/runtime. Copy the prefix from sibling YAML; do not invent it.
• Custom operators live in Python (plugins/operators/ or equivalent). YAML only references them. New reusable logic → operator + YAML task, not a one-off Python DAG.
• Python callables in YAML: python_callable_file + python_callable_name (absolute path inside the image, usually /usr/local/airflow/dags/...).
• max_active_runs: 1 means a manual trigger can sit queued behind a scheduled run.

5. Validate

Prefer parse over hoping the UI will show the DAG:

astro dev parse

If the repo wraps Airflow CLI:

# only if these targets exist
make dags-import-errors
make airflow CMD='dags list'

Broken YAML = import error, not a runtime task failure. Fix the YAML/loader, then re-parse.

PyPI v1 also ships dagfactory convert (Airflow 2 YAML → 3 operator paths). Do not run convert on map-format in-repo YAML.

6. New DAG checklist

1. Read one sibling YAML in the same source folder and copy structure.
2. DAG id: [job_type]_[source]_[object] if the repo already uses that pattern (elt_, mon_, mnt_, rpt_, ...).
3. Loader dag.py in the source folder; YAML under configs/.
4. Both callback layers + catchup: false unless backfill is the point.
5. astro dev parse clean before claiming done.

7. Troubleshooting

Related

• Remote runs/logs: vd:astro-airflow
• Local astro dev: vd:managing-astro-local-env
• Otto: vd:delegating-to-otto
• Upstream v1 reference: https://github.com/astronomer/dag-factory
• Official skill (list-format): https://github.com/astronomer/agents/blob/main/skills/dag-factory/SKILL.md