agentsclimarketplace

Airflow dag factory

Skill ivanshamaev/de-agent-skills/skills/airflow_dag_factory

Профессиональные Data Engineering Agent Skills для разработки AI Agentic Data Platform

Install
npx -y skills add ivanshamaev/de-agent-skills --skill airflow_dag_factory

Assembled from the repository path, not quoted from the project. Check it against their README if it does not work.

2 things to look at

  • no licenseNo license file was found in the repository. Code published without one is not open source by default, so using it at work is a question for whoever answers licensing questions where you are.
  • 13 stars13 stars. Stars are a popularity signal and not a quality one, but at this level it is likely that nobody has read this closely except its author, and you would be relying on your own review.

What its author says it does

Copied from the file, not written here

Use when building, reviewing, debugging, or scaling Apache Airflow DAGs declaratively with dag-factory YAML configs — including project setup, loader configuration, defaults hierarchy, custom/provider operators, dynamic task mapping, datasets/assets, callbacks, TaskFlow decorators, Jinja2 templating, environment variables, YAML DRY patterns, large-scale multi-DAG generation, CI/CD linting, and migration from pre-1.0.

SKILL.md

29.8 KB, as published. Nobody here has run it

Airflow DAG Factory

Build Apache Airflow DAGs declaratively with dag-factory — a library that turns YAML configuration files into Airflow DAGs with no boilerplate Python.

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+, Python 3.10+, Airflow 2.9+ (Airflow 3 supported)
For pre-1.0 projects, read reference/migration.md before applying any guidance here.


When to Use dag-factory

ScenarioRecommendation
Low-code, repetitive DAG patterns (same operators, varying params)dag-factory — ideal
Many similar DAGs that differ only in config valuesdag-factory — generate many DAGs from one YAML
Complex branching, dynamic Python logic, Pydantic validationPython DAGs or blueprint skill instead
Full TaskFlow pipeline in Pythonauthoring-dags skill instead
Existing dag-factory <1.0 projectFollow reference/migration.md first

Quick Decision Table

User RequestGo to
"Create a YAML DAG" / "Convert Python DAG to YAML"Defining a DAG in YAML
"Set up dag-factory in my project"Project Setup
"Share defaults across DAGs / set start_date once"Defaults
"Custom operator / KPO / Slack / Snowflake"Custom & Provider Operators
"Dynamic / mapped tasks / expand / partial"Dynamic Task Mapping
"Schedule on dataset / outlets / inlets"Datasets and Assets
"Add callback / Slack on failure"Callbacks
"Use @task decorator / TaskFlow in YAML"TaskFlow API
"Use env vars in YAML"Environment Variables
"Jinja2 templates in YAML"Jinja2 Templating
"Generate 50 DAGs from one template"Large-Scale Patterns
"DRY config / reuse blocks across DAGs"DRY with YAML Anchors
"Lint / validate YAML"Validation Commands
"Migrate from Airflow 2 → 3"Validation Commands (dagfactory convert)
"Migrate from dag-factory <1.0"reference/migration.md
Errors / broken DAGTroubleshooting

Project Setup

1. Install

Add to requirements.txt:

dag-factory>=1.0.0

dag-factory does not install Airflow providers. Install providers your YAML references explicitly:

apache-airflow-providers-slack
apache-airflow-providers-cncf-kubernetes
apache-airflow-providers-http

2. Create the Loader File

Create dags/load_dags.py. Airflow discovers DAGs from this file:

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 (recommended)
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 a Python dict (useful for testing)
# load_yaml_dags(globals_dict=globals(), config_dict={...})

globals_dict=globals() is required — generated DAG objects must be registered into the module namespace for Airflow to discover them.

3. Recommended Project Structure

project/
├── dags/
│   ├── load_dags.py          # single loader — only Python file in dags/
│   ├── defaults.yml           # global defaults (schedule, owner, retries, start_date)
│   ├── ingestion/
│   │   ├── defaults.yml       # team-level overrides
│   │   ├── orders.yml
│   │   └── customers.yml
│   └── transformation/
│       ├── defaults.yml
│       └── mart_revenue.yml
├── include/
│   ├── callbacks/
│   │   └── slack.py           # callback functions
│   └── tasks/
│       └── transforms.py      # Python callables for PythonOperator / @task
└── requirements.txt

Key rules:

  • Keep dags/ clean — only the loader .py and YAML/YML config files.
  • Put all Python callables, SQL, and helpers in include/ — the DAG processor ignores this directory by default.
  • Never put defaults.yml logic directly in dags_folder root if you point dags_folder at that directory — dag-factory may try to parse it as a DAG config. Use defaults_config_path to separate it, or use a default: block inside each YAML file.

4. Verify

dagfactory --version
dagfactory lint dags/

Defining a DAG in YAML

Each top-level YAML key (other than default) defines one DAG. The key becomes the dag_id.

# dags/ingestion/orders.yml
default:
  default_args:
    start_date: 2025-01-01
    owner: data-team
    retries: 2
  catchup: false
  schedule: "0 3 * * *"

ingest_orders:
  description: "Load orders from source to staging"
  tags: [ingestion, orders]
  task_groups:
    - group_name: validate
      tooltip: "Data quality checks"
      dependencies: [extract]
  tasks:
    - task_id: extract
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo extracting orders"
    - task_id: load
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo loading orders"
      dependencies: [extract]
    - task_id: check_row_count
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo checking row count"
      dependencies: [load]
      task_group_name: validate
    - task_id: check_nulls
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo checking nulls"
      dependencies: [load]
      task_group_name: validate

Key Fields Reference

FieldLevelDescription
defaultfileShared args applied to every DAG in this file
default_argsDAG or defaultStandard Airflow default_args (owner, retries, start_date, …)
scheduleDAGCron string, preset (@daily), Dataset URI list, or __type__ timetable
catchupDAGBoolean; default Airflow behavior is true — set false explicitly
descriptionDAGHuman-readable description
tagsDAGList of tag strings
max_active_runsDAGMax concurrent DAG runs
dagrun_timeoutDAG__type__: datetime.timedelta block
tasksDAGList of task dicts; each needs task_id + operator
operatortaskFull Python import path to operator class
dependenciestask/groupList of upstream task_ids or group_names
task_groupsDAGList of group dicts; each needs group_name
task_group_nametaskAssigns task to a task group

Tasks do not need to be listed in dependency order — dag-factory resolves topology from dependencies.

List vs Dict Format

Use the list format (shown above) for all new code — it is the v1.0 standard and required for task groups.

The legacy dict format (where tasks is a dict keyed by task_id) still works for backward compatibility but is not recommended.


Defaults

Four layers of defaults, in precedence order (highest wins):

PrioritySourceHow
1 (highest)Individual DAG blockdefault_args: or DAG-level keys inside the DAG
2default: block in the same YAML fileTop-level default: key
3defaults_config_dict=Passed as Python dict to load_yaml_dags
4 (lowest)defaults.yml fileAuto-detected walking up the dir tree, or explicit defaults_config_path=

default: Block (same file)

Best for templates with multiple DAGs in one file:

default:
  default_args:
    owner: analytics
    start_date: 2025-01-01
    retries: 1
    retry_delay:
      __type__: datetime.timedelta
      minutes: 5
  catchup: false
  schedule: "@daily"
  tags: [analytics]

revenue_daily:
  description: "Daily revenue mart"
  tasks:
    - task_id: build
      operator: airflow.operators.bash.BashOperator
      bash_command: "dbt run --select mart_revenue"

users_daily:
  description: "Daily user mart"
  tasks:
    - task_id: build
      operator: airflow.operators.bash.BashOperator
      bash_command: "dbt run --select mart_users"

defaults.yml File (hierarchical, team-wide)

Place defaults.yml files in the directory tree. dag-factory merges them walking from the DAG file up to the root, with files closer to the DAG winning:

# dags/defaults.yml — global
schedule: "0 6 * * *"
catchup: false
default_args:
  start_date: '2025-01-01'
  owner: platform
  retries: 2
# dags/ingestion/defaults.yml — team-level overrides
default_args:
  owner: ingestion-team
  retries: 3
tags: [ingestion]

A DAG inside dags/ingestion/ inherits both files, with ingestion/defaults.yml values winning.

Important: set defaults_config_path in the loader to the highest ancestor folder you want merged:

load_yaml_dags(
    globals_dict=globals(),
    dags_folder="/usr/local/airflow/dags/",
    defaults_config_path="/usr/local/airflow/dags/",  # walk from here
)

Custom & Provider Operators

Reference any operator by its full Python import path. All other task keys become __init__ kwargs.

tasks:
  - task_id: wait_for_api
    operator: airflow.providers.http.sensors.http.HttpSensor
    http_conn_id: api_conn
    endpoint: /health
    request_params: {}
    response_check:
      __type__: builtins.str
      __args__: ["lambda response: response.json()['status'] == 'ok'"]
    poke_interval: 30
    timeout: 600
    mode: reschedule

  - task_id: run_snowflake
    operator: airflow.providers.snowflake.operators.snowflake.SnowflakeOperator
    snowflake_conn_id: snowflake_default
    sql: "CALL my_procedure()"
    warehouse: TRANSFORM_WH
    database: ANALYTICS
    schema: PUBLIC

  - task_id: custom_op
    operator: include.operators.MyCustomOperator
    my_param: "value"

For Airflow 3, prefer airflow.providers.standard.operators.* over the legacy airflow.operators.* — use dagfactory convert to auto-rewrite:

dagfactory convert dags/ --override

KubernetesPodOperator

Use __type__ for nested k8s objects (legacy auto-casting was removed in v1.0):

tasks:
  - task_id: run_pod
    operator: airflow.providers.cncf.kubernetes.operators.pod.KubernetesPodOperator
    image: python:3.12-slim
    cmds: ["python", "-c"]
    arguments: ["print('hello from pod')"]
    name: my-pod
    namespace: default
    get_logs: true
    is_delete_operator_pod: true
    container_resources:
      __type__: kubernetes.client.models.V1ResourceRequirements
      limits:
        cpu: "1"
        memory: "1Gi"
      requests:
        cpu: "500m"
        memory: "512Mi"
    env_vars:
      - __type__: kubernetes.client.models.V1EnvVar
        name: MY_ENV
        value: "my_value"

Dynamic Task Mapping

Use expand and partial keys. Two reference syntaxes — do not mix them:

SyntaxUse case
task_id.outputInside expand.op_args / expand.op_kwargs (XCom-style)
+task_idAs a bare expand value; for TaskFlow decorator tasks

PythonOperator — XCom-style expand

tasks:
  - task_id: get_ids
    operator: airflow.providers.standard.operators.python.PythonOperator
    python_callable_name: fetch_ids
    python_callable_file: /usr/local/airflow/include/tasks/extract.py

  - task_id: process_id
    operator: airflow.providers.standard.operators.python.PythonOperator
    python_callable_name: process_one
    python_callable_file: /usr/local/airflow/include/tasks/extract.py
    partial:
      op_kwargs:
        env: production
    expand:
      op_args: get_ids.output          # XCom-style: inside op_args
    dependencies: [get_ids]

TaskFlow — bare +task_id expand

tasks:
  - task_id: build_numbers
    decorator: airflow.sdk.definitions.decorators.task
    python_callable: include.tasks.sample.build_numbers_list

  - task_id: double
    decorator: airflow.sdk.definitions.decorators.task
    python_callable: include.tasks.sample.double_number
    expand:
      number: +build_numbers           # bare +: expand over XComArg
    dependencies: [build_numbers]

map_index_template (Airflow 2.9+)

  - task_id: process_item
    operator: airflow.providers.standard.operators.python.PythonOperator
    python_callable_name: process
    python_callable_file: /usr/local/airflow/include/tasks/process.py
    expand:
      op_args: get_items.output
    map_index_template: "{{ task.item_id }}"   # callable sets context["item_id"]

Supported patterns: simple mapping, task-generated mapping, partial, multiple-parameter mapping, map_index_template.
Unsupported: mapping over task groups, zipping expanding collections, transform().


Datasets and Assets

Producer / Consumer pattern

ingest_orders:
  default_args:
    start_date: '2025-01-01'
  schedule: "0 3 * * *"
  catchup: false
  tasks:
    - task_id: load
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo loading"
      outlets:
        - s3://datalake/orders/latest.parquet

transform_orders:
  default_args:
    start_date: '2025-01-01'
  schedule:
    - s3://datalake/orders/latest.parquet   # triggered when producer updates this
  catchup: false
  tasks:
    - task_id: transform
      operator: airflow.operators.bash.BashOperator
      bash_command: "echo transforming"

Conditional Dataset Scheduling (Airflow 2.9+ / dag-factory 0.22+)

schedule:
  datasets:
    __or__:
      - __and__:
          - s3://datalake/orders/latest.parquet
          - s3://datalake/customers/latest.parquet
      - s3://datalake/orders_fallback/latest.parquet

Airflow 3 gotcha: task-level outlets/inlets may be silently ignored for asset-aware scheduling in some Airflow 3.x versions. If asset-triggered DAGs don't fire, verify with a Python DAG first.


Callbacks

Three styles — all valid at DAG, TaskGroup, or Task level (or under default_args):

1. String callable path (simplest)

tasks:
  - task_id: extract
    operator: airflow.operators.bash.BashOperator
    bash_command: "extract.sh"
    on_failure_callback: include.callbacks.slack.notify_failure
    on_success_callback: include.callbacks.metrics.record_success

2. Callable with kwargs

tasks:
  - task_id: load
    operator: airflow.operators.bash.BashOperator
    bash_command: "load.sh"
    on_failure_callback:
      callback: include.callbacks.slack.send_message
      channel: "#data-alerts"
      mention: "@oncall"

3. File path + function name

tasks:
  - task_id: validate
    operator: airflow.operators.bash.BashOperator
    bash_command: "validate.sh"
    on_retry_callback_name: log_retry
    on_retry_callback_file: /usr/local/airflow/include/callbacks/logging.py

4. Provider callback (e.g. Slack)

tasks:
  - task_id: critical_load
    operator: airflow.operators.bash.BashOperator
    bash_command: "load.sh"
    on_failure_callback:
      callback: airflow.providers.slack.notifications.slack.send_slack_notification
      slack_conn_id: slack_default
      text: ":red_circle: *{{ dag.dag_id }}* / *{{ task.task_id }}* failed."
      channel: "#data-oncall"

TaskFlow API

Use the decorator key with the full import path to the Airflow task decorator. Reference the callable via python_callable: module.function.

pipeline_dag:
  default_args:
    start_date: 2025-01-01
  schedule: "@daily"
  tasks:
    - task_id: extract
      decorator: airflow.sdk.definitions.decorators.task
      python_callable: include.tasks.pipeline.extract_data

    - task_id: transform
      decorator: airflow.sdk.definitions.decorators.task
      python_callable: include.tasks.pipeline.transform_data
      dependencies: [extract]

    - task_id: load
      decorator: airflow.sdk.definitions.decorators.task
      python_callable: include.tasks.pipeline.load_data
      dependencies: [transform]

Corresponding callable file (include/tasks/pipeline.py):

def extract_data() -> list[dict]:
    return [{"id": 1}, {"id": 2}]

def transform_data(records: list[dict]) -> list[dict]:
    return [{"id": r["id"], "processed": True} for r in records]

def load_data(records: list[dict]) -> None:
    print(f"Loading {len(records)} records")

Mix TaskFlow tasks with traditional operator tasks freely — they resolve via dependencies like any other task.


Environment Variables

dag-factory supports os.environ expansion in several places. Use env vars for environment-specific config (paths, connection IDs, bucket names):

In the loader

import os
from dagfactory import load_yaml_dags

load_yaml_dags(
    globals_dict=globals(),
    dags_folder=os.getenv("DAGS_FOLDER", "/usr/local/airflow/dags/"),
    defaults_config_path=os.getenv("DAGS_FOLDER", "/usr/local/airflow/dags/"),
)

In YAML via $VAR / ${VAR} substitution

dag-factory expands environment variables in YAML string values:

tasks:
  - task_id: run_script
    operator: airflow.providers.standard.operators.python.PythonOperator
    python_callable_name: run
    python_callable_file: $CONFIG_ROOT_DIR/include/tasks/run.py  # expanded at load time

  - task_id: load_s3
    operator: airflow.operators.bash.BashOperator
    bash_command: "aws s3 cp data.csv s3://$S3_BUCKET/raw/"

Set env vars in Dockerfile, .env, Airflow connection settings, or Astro CLI airflow_settings.yaml.


Jinja2 Templating

Standard Airflow Jinja2 macros work in any templated operator field:

tasks:
  - task_id: extract_partition
    operator: airflow.operators.bash.BashOperator
    bash_command: >
      python extract.py
        --date {{ ds }}
        --start {{ data_interval_start }}
        --end {{ data_interval_end }}

  - task_id: run_sql
    operator: airflow.providers.snowflake.operators.snowflake.SnowflakeOperator
    snowflake_conn_id: snowflake_default
    sql: |
      SELECT * FROM orders
      WHERE order_date = '{{ ds }}'
        AND run_id = '{{ run_id }}'

Useful macros: {{ ds }}, {{ ds_nodash }}, {{ ts }}, {{ run_id }}, {{ dag.dag_id }}, {{ task.task_id }}, {{ data_interval_start }}, {{ data_interval_end }}, {{ var.value.my_var }}, {{ conn.my_conn_id.host }}.


Custom Python Objects (__type__)

For any non-scalar value (datetime, timedelta, timetable, k8s objects), use __type__:

# Timeout with timedelta
execution_timeout:
  __type__: datetime.timedelta
  hours: 2
  minutes: 30

# start_date as datetime
start_date:
  __type__: datetime.datetime
  year: 2025
  month: 1
  day: 1

# Custom timetable
schedule:
  __type__: airflow.timetables.trigger.CronTriggerTimetable
  cron: "0 1 * * 1-5"
  timezone: Europe/Moscow

# List of typed objects
env_vars:
  __type__: builtins.list
  items:
    - __type__: kubernetes.client.models.V1EnvVar
      name: ENVIRONMENT
      value: production
    - __type__: kubernetes.client.models.V1EnvVar
      name: LOG_LEVEL
      value: INFO

__type__ is the full import path to the class.
__args__ is a list of positional constructor arguments.
Other keys become keyword arguments.

Reserved keys — do not use for your own data: __type__, __args__, __join__, __and__, __or__, and items inside a __type__: builtins.list block.


DRY with YAML Anchors

YAML native anchors and merge keys (<<) reduce repetition across tasks in one file without any dag-factory-specific features:

# Define reusable blocks with & — place at top of file
x-python-defaults: &python-defaults
  operator: airflow.providers.standard.operators.python.PythonOperator
  python_callable_file: /usr/local/airflow/include/tasks/etl.py

x-retry-policy: &retry-policy
  retries: 3
  retry_delay:
    __type__: datetime.timedelta
    minutes: 10
  execution_timeout:
    __type__: datetime.timedelta
    hours: 1

default:
  default_args:
    <<: *retry-policy
    owner: data-team
    start_date: 2025-01-01
  catchup: false

ingest_orders:
  schedule: "0 2 * * *"
  tasks:
    - task_id: extract
      <<: *python-defaults
      python_callable_name: extract_orders
    - task_id: validate
      <<: *python-defaults
      python_callable_name: validate_orders
      dependencies: [extract]

ingest_customers:
  schedule: "0 3 * * *"
  tasks:
    - task_id: extract
      <<: *python-defaults
      python_callable_name: extract_customers
    - task_id: validate
      <<: *python-defaults
      python_callable_name: validate_customers
      dependencies: [extract]

Rules:

  • Define anchors at the top of the file.
  • Name anchors descriptively (&retry-policy, not &a1).
  • Use x- prefix for anchor-only blocks (the x- prefix is ignored by YAML parsers as an extension field — it will not become a DAG).
  • Avoid deep nesting; merge only what actually varies.

Large-Scale Patterns

Many DAGs from a single file

One YAML file can define dozens of DAGs. Use default: for shared config, anchors for shared task structure:

default:
  default_args:
    start_date: 2025-01-01
    owner: ingestion
    retries: 2
  catchup: false
  schedule: "0 4 * * *"

x-ingest-tasks: &ingest-tasks
  tasks:
    - task_id: extract
      operator: airflow.operators.bash.BashOperator
      bash_command: "python include/ingest.py --source {{ dag.dag_id }}"
    - task_id: load
      operator: airflow.operators.bash.BashOperator
      bash_command: "python include/load.py --source {{ dag.dag_id }}"
      dependencies: [extract]

ingest_orders:
  <<: *ingest-tasks
  tags: [orders]

ingest_customers:
  <<: *ingest-tasks
  tags: [customers]

ingest_products:
  <<: *ingest-tasks
  tags: [products]
  schedule: "0 5 * * *"   # override default schedule for this DAG

Programmatic YAML generation

For truly large fleets (50+ DAGs), generate YAML files from a template rather than hand-writing them. Run the generator in CI or as a pre-deploy step:

# scripts/generate_dags.py
import yaml
from pathlib import Path

SOURCES = [
    {"name": "orders",    "schedule": "0 2 * * *", "table": "raw.orders"},
    {"name": "customers", "schedule": "0 3 * * *", "table": "raw.customers"},
    # ... more sources
]

template = {
    "default": {
        "default_args": {"start_date": "2025-01-01", "retries": 2},
        "catchup": False,
    }
}

for source in SOURCES:
    dag_config = {
        f"ingest_{source['name']}": {
            "schedule": source["schedule"],
            "tasks": [
                {
                    "task_id": "extract",
                    "operator": "airflow.operators.bash.BashOperator",
                    "bash_command": f"python include/ingest.py --table {source['table']}",
                }
            ],
        }
    }
    template.update(dag_config)

output = Path("dags/generated/ingestion.yml")
output.write_text(yaml.dump(template, sort_keys=False))

Symlink-based partitioning (for very large fleets)

When a single YAML generates hundreds of DAGs and Airflow DAG processor overhead becomes visible, split into symlinked loader files:

# dags/load_orders.py  (symlink to dags/load_factory.py)
# dags/load_customers.py  (symlink to dags/load_factory.py)

# dags/load_factory.py
import os
from dagfactory import load_yaml_dags

# Each symlink has a different __file__ name
source = os.path.basename(__file__).replace("load_", "").replace(".py", "")
load_yaml_dags(
    globals_dict=globals(),
    config_filepath=f"/usr/local/airflow/dags/generated/{source}.yml",
)

Each symlink processes only its own YAML, keeping the Airflow scheduler's DAG parse scope smaller.


Validation Commands

# Verify install
dagfactory --version

# Lint YAML syntax for a file or directory
dagfactory lint dags/

# Verbose: per-file results table
dagfactory lint dags/ --verbose

# Show diffs to migrate Airflow 2 → 3 import paths
dagfactory convert dags/

# Apply conversions in place
dagfactory convert dags/ --override

dagfactory lint validates YAML syntax only. Operator import errors and missing kwargs surface at Airflow parse time.

Full validation workflow

# 1. YAML syntax
dagfactory lint dags/

# 2. Airflow 2 → 3 migration (if needed)
dagfactory convert dags/ --override

# 3. Airflow parse (Astro CLI)
astro dev parse

# 4. Alternative: bare Airflow
airflow dags list-import-errors

CI/CD Integration

GitHub Actions snippet

# .github/workflows/validate_dags.yml
name: Validate DAGs
on: [push, pull_request]

jobs:
  lint-and-parse:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: pip install dag-factory>=1.0.0 apache-airflow>=2.9

      - name: Lint YAML
        run: dagfactory lint dags/ --verbose

      - name: Parse DAGs
        run: airflow dags list-import-errors --output json | python -c "
import sys, json
errors = json.load(sys.stdin)
if errors: sys.exit(1)
"

Pre-commit hook

# .pre-commit-config.yaml
repos:
  - repo: local
    hooks:
      - id: dagfactory-lint
        name: dag-factory lint
        entry: dagfactory lint
        language: python
        files: \.ya?ml$
        pass_filenames: true

Common Pitfalls

PitfallRoot causeFix
DAG doesn't appear in UIglobals_dict=globals() not passed to load_yaml_dagsAdd globals_dict=globals()
defaults.yml shows as broken DAGFile in dags_folder without a valid DAG keyUse defaults_config_path= to point at its parent; name it outside dags_folder
ModuleNotFoundError on operatorProvider not installedAdd apache-airflow-providers-<name> to requirements.txt
ModuleNotFoundError after Airflow 3 upgradeLegacy airflow.operators.* paths removedRun dagfactory convert dags/ --override
Wrong type on scalar (e.g. start_date)String passed where datetime expectedUse __type__: datetime.datetime block
XCom reference not working in expandMixed +task_id and task_id.output syntaxUse task_id.output inside op_args/op_kwargs; use +task_id for bare expand
Asset outlets silently ignored on Airflow 3.xKnown issue #718Verify with a Python DAG; track issue on GitHub
dagfactory tag added to all DAGsv1.0 auto-adds itFilter by tag in UI, or override with your own tags: list
Conditional dataset schedule ignoredAirflow <2.9 or dag-factory <0.22Upgrade; rename !and/!or__and__/__or__
Multiple defaults.yml not mergingdefaults_config_path not pointing at root ancestorSet it to the highest dir you want included
KPO nested objects errorLegacy auto-casting removed in v1.0Use __type__: kubernetes.client.models.V1...

Troubleshooting

DAG missing from Airflow UI

  1. Check dags/load_dags.py exists and calls load_yaml_dags(globals_dict=globals(), ...).
  2. Run dagfactory lint dags/ — fix any YAML syntax errors.
  3. Run astro dev parse or airflow dags list-import-errors.
  4. Check the Airflow webserver and scheduler logs for parse errors.

Import errors

ModuleNotFoundError: No module named 'airflow.operators.email'

For Airflow 3: run dagfactory convert dags/ --override. Then reinstall: pip install apache-airflow-providers-standard.

YAML parses but tasks don't run / wrong topology

  • Verify dependencies: lists correct task_id values (case-sensitive).
  • For task groups, task_group_name: on a task must match group_name: on the group definition.
  • Check dagfactory lint — it catches structural YAML errors.

XCom mapping not working

  • Inside expand.op_args → use task_id.output syntax.
  • As a bare expand value → use +task_id syntax.
  • Never mix the two in the same task.

Timedelta / datetime type errors

Replace:

# BAD — string will fail
execution_timeout: "3600"
retry_delay: "00:05:00"

With:

# GOOD — __type__ syntax
execution_timeout:
  __type__: datetime.timedelta
  hours: 1
retry_delay:
  __type__: datetime.timedelta
  minutes: 5

Verification Checklist

Before calling a DAG complete:

  • dagfactory lint dags/ passes with no errors
  • astro dev parse (or airflow dags list-import-errors) shows no import errors
  • DAG appears in Airflow UI with correct dag_id, schedule, and tags
  • All task dependencies render correctly in Grid/Graph view
  • Required provider packages listed in requirements.txt
  • Python callables are in include/, not in dags/
  • defaults.yml not in dags_folder root (or defaults_config_path set correctly)

Related Skills

  • authoring-dags — Pure Python Airflow DAGs with af CLI. Use when YAML cannot express the logic.
  • testing-dags — Testing DAGs, debugging failures, test → fix → retest loop.
  • debugging-dags — Troubleshooting failed DAG runs.

References

Keep looking

Skills are one crate of 328,083. Ordering is by how many stacks a row turns up in, so the top of any crate is what has actually been picked rather than what has the most stars.