Use DAG Factory to create dags

Learn how to dynamically convert YAML files into Apache Airflow® dags with DAG Factory, an open source project that makes creating dags easy.

DAG Factory is an open source tool managed by Astronomer that allows you to dynamically generate Apache Airflow® dags from YAML. While Airflow dags are traditionally written exclusively in Python, DAG Factory makes it easy for people who don’t know Python to use Airflow.

This guide provides a complete walkthrough of using DAG Factory package to build production-ready pipelines in a modern Airflow project. You will learn to install the library, structure your project according to best practices, and define a multi-task pipeline entirely in YAML. The example demonstrates powerful features like using the TaskFlow API, organizing tasks with task groups, and passing data between tasks, all from your configuration file. By the end, you’ll be ready to apply these patterns to your own dynamic dags.

DAG Factory can be used with all Astronomer products and any Apache Airflow installation. To view the source code of the project, have a look at the dag-factory GitHub repository.

When to Use DAG Factory

While writing dags directly in Python is powerful and flexible, it’s not always the most efficient approach for every use case. DAG Factory offers a configuration-driven alternative where you define the structure of your pipelines in YAML. This is particularly useful in several key scenarios:

Empowering Teams

YAML is often more approachable than Python. DAG Factory allows team members like analysts or junior engineers, who may not be Airflow experts, to create and manage their own dags with a simplified, declarative syntax.

Standardizing Repetitive Dags

If you have dozens of dags that follow the same pattern (like a standard extract-and-load job), DAG Factory is ideal. You can create a standard template and then generate numerous dags just by changing the parameters in a YAML file, which reduces code duplication and simplifies maintenance.

Separating Logic from Structure

DAG Factory helps you separate the what from the how. The YAML clearly defines the dag’s structure and dependencies, while the underlying Python functions handle the actual business logic. This makes your dags easier to read at a glance and your Python code more modular and testable.

While DAG Factory offers significant advantages for some use cases, there are scenarios when using other ways of dag authoring like writing dags directly in Python are more appropriate.

When your data pipelines require complex conditional logic, branching or sophisticated error handling that goes beyond what YAML can express cleanly, native Python is generally the better approach. Additionally, YAML-based dags can be more challenging to debug compared to native Python code, as you are missing extensive logging or step-through debugging capabilities. Finally, your way of orchestrating workflows should match your team environment, so consider existing expertise.

While DAG Factory is a flexible product that supports all the main concepts of Airflow, newer features like asset-aware scheduling may work but are not as user-friendly or as well integrated as others.

Assumed knowledge

To get the most out of this tutorial, you should have an understanding of:

• The Airflow components and how they work together.
• Airflow fundamentals, such as writing dags and defining tasks.
• Basic understanding of Airflow operators.

Prerequisites

• Python 3.9.0+
• The Astro CLI

Step 1: Initialize your Airflow Project with the Astro CLI

First, create a new project directory and initialize an Astro project using the Astro CLI.

mkdir my-dag-factory-project && cd my-dag-factory-project
astro dev init

The init command creates a standard Airflow project structure. Since this tutorial focuses on DAG Factory, let’s remove the example dag that’s included by default.

rm dags/exampledag.py

Next, add the dag-factory library as a project dependency. Open requirements.txt and add the following line:

dag-factory==1.0.1

Now, start your local Airflow environment. The Astro CLI will build your project, installing dag-factory in the process.

astro dev start

Once the project is running, the Airflow UI should open automatically at http://localhost:8080 and you will be presented with an empty dags list.

Step 2: Organizing the project

A key to building a maintainable and performant Airflow project is proper organization. While you could put all your YAML configs, Python scripts, and SQL files into the dags/ folder, this can quickly become messy and put unnecessary strain on the dag processor.

Astronomer recommends placing Python, SQL, and other scripts that are not dag definitions in the include/ folder. Files in this folder are available to your dags but are not parsed by the Airflow dag processor, which reduces overhead and improves performance.

For this tutorial, we’ll use a structure that is also a great starting point for real-world projects:

• dags/: This folder will contain only the YAML configuration files and the Python script that generates the dags from them. This keeps all dag definitions encapsulated.
• include/: We will create a tasks subfolder here to hold the Python functions that our operators will call. Any other supporting scripts (e.g. SQL queries) would also live in sub-folders within include/.

We will apply this principle in the next steps.

For larger projects with a mix of dynamically generated and standard Python dags, consider organizing further. For example, you could create a dags/configs subfolder to hold all your DAG Factory YAML files, keeping them separate from your other .py dag files.

To separate our business logic from the rest of our orchestration logic, create a new folder named tasks inside include. There, we’ll add Python scripts that define the functions our YAML-based pipelines will call in the next steps.

mkdir -p include/tasks

Step 3: Prepare functions

Our example dag will orchestrate a simple pipeline using both a PythonOperator using the TaskFlow API and a BashOperator using the traditional operator.

DAG Factory supports both traditional operators and the modern TaskFlow API. This tutorial targets Airflow 3.x and will use the TaskFlow API decorator syntax whenever possible.

Before defining the dag in YAML, let’s write the Python functions that our tasks will execute. Following our plan from Step 2, we’ll place these functions in the include/tasks/ folder.

Create a file named include/tasks/basic_example_tasks.py with the following content:

def _extract_data() -> list[int]:
    return [1, 2, 3, 4]

def _store_data(processed_at: str, data_a: list[int], data_b: list[int]) -> None:
    print(f"Storing {len(data_a + data_b)} records at {processed_at}")

Design your Python functions to be small, self-contained, and independently testable, which aligns with best practices for both DAG Factory and general Airflow development.

Step 4: Define a basic dag in YAML

Now we can create the YAML definition for our dag. Create a new YAML file in the dags folder named basic_example.yml and add the following content:

basic_example_dag:
  default_args:
    owner: "astronomer"
    start_date: 2025-09-01
  description: "Basic example DAG"
  tags: ["demo", "etl"]
  schedule: "@hourly"

  task_groups:
    extract:
      tooltip: "data extraction"

  tasks:
    extract_data_from_a:
      decorator: airflow.sdk.task
      python_callable: include.tasks.basic_example_tasks._extract_data
      task_group_name: extract

    extract_data_from_b:
      decorator: airflow.sdk.task
      python_callable: include.tasks.basic_example_tasks._extract_data
      task_group_name: extract

    store_data:
      decorator: airflow.sdk.task
      python_callable: include.tasks.basic_example_tasks._store_data
      processed_at: "{{ logical_date }}"
      data_a: +extract_data_from_a
      data_b: +extract_data_from_b
      dependencies: [extract]

    validate_data:
      operator: airflow.providers.standard.operators.bash.BashOperator
      bash_command: "echo data is valid"
      dependencies: [store_data]

This YAML file defines the dag’s structure and its tasks. Note how +extract_data_from_a and +extract_data_from_b are used to pass the return value of the extract tasks to the store_data task, and how Jinja templating ({{ logical_date }}) is used to pass the logical date.

Step 5: Implement the generator script

The final step to make our dag appear, is to create the Python script that Airflow will parse. This script uses the DAG Factory library to find our YAML file and generate the actual Airflow dag object from it. This approach gives you full control over the generation process and allows for extensive customization in advanced use cases.

Create a Python file named dags/basic_example_dag_generation.py with the following content:

import os
from pathlib import Path

from dagfactory import load_yaml_dags

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

config_file = str(CONFIG_ROOT_DIR / "basic_example.yml")

load_yaml_dags(
    globals_dict=globals(),
    config_filepath=config_file,
)

Once the dag processor parses this file, your dag with the ID basic_example_dag will appear in the UI. It has 4 tasks in its pipeline, 2 of them within a task group:

• extract_data: Uses the TaskFlow API to call the _extract_data function from our include/tasks/basic_example_tasks.py script. We create 2 different tasks in this scenario. Both return a list of numbers.
• store_data: Uses the TaskFlow API to call the _store_data function from our include/tasks/basic_example_tasks.py script. As you can see, to pass parameters with this approach, just set them with the appropriate name directly in the YAML configuration. With +extract_data we tell DAG Factory to reference the return value of the extract_data task. Also as shown in the example, you can use Jinja templating including variables, macros and filters.
• validate_data: Here, we use the classic approach to use the BashOperator, just printing the sentence data is valid.

The load_yaml_dags function is responsible for generating the dags. You can point it to a specific file, or to a folder that it will scan recursively for .yml or .yaml files. It uses the provided globals_dict to add the generated dags to the Airflow context. For more options, see the official documentation.

With this, you already know the basics of how to orchestrate a dag with YAML, including using task groups, passing data between tasks, using the TaskFlow API and classic operators, and setting basic dag attributes like the schedule or tags.

Dags defined with DAG Factory automatically receive the dagfactory tag. Also, if you select Dag Docs from the individual dag view, it will by default show the YAML file that created the dag, which is very useful for debugging.

This is a great starting point, and the following steps will cover more advanced features to prepare your DAG Factory knowledge for real-world use cases.

(Optional) Step 6: Asset-Aware Scheduling with YAML

Now, let’s explore one of Airflow’s most powerful features, asset-aware scheduling, and how to implement it using DAG Factory. We will create two dags: a producer that updates an asset, and a consumer that runs whenever that asset is updated.

First, let’s create the Python functions that our tasks will execute. These functions will fetch data from an API, save it to a file, and then read it back.

Create a new file named include/tasks/asset_example_tasks.py with the following content:

`include/tasks/asset_example_tasks.py`

import json
import tempfile
import requests

def _get_iss_coordinates_file_path() -> str:
    return tempfile.gettempdir() + "/iss_coordinates.txt"

def _update_iss_coordinates() -> None:
    placeholder = {"latitude": "0.0", "longitude": "0.0"}

    try:
        response = requests.get("http://api.open-notify.org/iss-now.json", timeout=5)
        response.raise_for_status()
        data = response.json()
        coordinates = data.get("iss_position", placeholder)
    except Exception:
        coordinates = placeholder

    with open(_get_iss_coordinates_file_path(), "w") as f:
        f.write(json.dumps(coordinates))

def _read_iss_coordinates() -> None:
    path = _get_iss_coordinates_file_path()
    with open(path, "r") as f:
        print("::group::ISS Coordinates")
        print(f.read())
        print("::endgroup::")

The _update_iss_coordinates function retrieves data from an API and writes it to a file, while _read_iss_coordinates reads this file and prints the content to a dedicated log group.

Now that we have our Python logic, we can define the two dags that will orchestrate it. Create a new YAML file at dags/asset_example.yml:

default:
  start_date: 2025-09-01

update_iss_coordinates:
  schedule: "@daily"
  tasks:
    update_coordinates:
      decorator: airflow.sdk.task
      python_callable: include.tasks.asset_example_tasks._update_iss_coordinates
      outlets:
        - __type__: airflow.sdk.Asset
          name: "iss_coordinates"

process_iss_coordinates:
  schedule:
    - __type__: airflow.sdk.Asset
      name: "iss_coordinates"
  tasks:
    read_coordinates:
      decorator: airflow.sdk.task
      python_callable: include.tasks.asset_example_tasks._read_iss_coordinates

This single YAML file defines both the update_iss_coordinates (producer) and process_iss_coordinates (consumer) dags. For the producing dag we define an outlet of type airflow.sdk.Asset and name it iss_coordinates. The consuming dag then uses this same asset identifier for its schedule attribute, which creates the dependency.

Also, take note of the YAML top-level default block. This configuration affects all the dags defined in the YAML file, allowing you to share standard settings and configurations, for improved consistency, maintainability and simplicity.

Finally, to generate these dags in Airflow, we need to create a corresponding generator script. Create a new file named dags/asset_example_dag_generation.py with the following content:

`dags/asset_example_dag_generation.py`

import os
from pathlib import Path

from dagfactory import load_yaml_dags

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

config_file = str(CONFIG_ROOT_DIR / "asset_example.yml")

load_yaml_dags(
    globals_dict=globals(),
    config_filepath=config_file,
)

And that’s it! Once the scheduler parses this file, you will see two new dags in the Airflow UI, connected by the iss_coordinates asset. When you run the update_iss_coordinates dag, the process_iss_coordinates dag will be triggered automatically upon its completion.

For a simpler approach to creating one dag with one task updating an asset, you could use @asset syntax, adding @asset(schedule="@daily") directly to the _update_iss_coordinates function in your Python file. This would allow you to remove the update_iss_coordinates dag definition from your YAML entirely. This tutorial defines both in YAML to fully demonstrate how DAG Factory handles asset producers and consumers.

(Optional) Step 7: Alternative YAML Loading

In the previous steps, we used a dedicated Python dag generation script for each dag to parse the YAML with DAG Factory. This is a useful approach for maximum control over the generation process, and to avoid any unexpected workload when teams work with many YAML files.

However, it also adds complexity. The load_yaml_dags function therefore also supports a more pragmatic way, to parse all YAML files in your dags folder recursively. To illustrate this, delete the two generator scripts dags/basic_example_dag_generation.py and dags/asset_example_dag_generation.py. Then create a new file dags/dag_generation.py:

# keep import to ensure the dag processor parses the file
from airflow.sdk import dag
from dagfactory import load_yaml_dags
load_yaml_dags(globals_dict=globals())

In this particular case, we need to add the dag import as an indicator for Airflow, to not skip this file during parsing. You will notice, the result will be the same as before, and all additional YAML files added will now be automatically processed.

When searching for dags inside the dag bundle, Airflow only considers Python files that contain the strings airflow and dag (case-insensitively) as an optimization. Because of these optimizations, you might need to add the dag import to ensure your file is parsed. To consider all Python files instead, disable the DAG_DISCOVERY_SAFE_MODE configuration flag.

In case you want to outsource your YAML definitions, you can overwrite the dags_folder argument when calling the load_yaml_dags function to set a custom folder to process recursively.

(Optional) Step 8: Configuration and inheritance

As you create more dags, you’ll want to avoid repeating the same configuration. DAG Factory includes powerful features for centralized configuration and inheritance to help you keep your dag definitions clean, consistent, and easy to maintain across your project.

This feature allows you to set default values for both dag-level arguments (like schedule) and task-level arguments (like retries via default_args).

In our dags/asset_example.yml file, you already discovered one way to configure dags in a centralized way within the YAML definition:

default:
  start_date: 2025-09-01

update_iss_coordinates:
  # ...

process_iss_coordinates:
  # ...

With this approach both dags, update_iss_coordinates and process_iss_coordinates, will use the start_date from the default block. This feature becomes even more powerful, when using global defaults in combination with inheritance.

To illustrate this, let’s imagine a real-world scenario with a set of company-wide data pipeline standards:

• All dags should have a default start_date of 2025-09-01.
• All dags should be owned by astronomer, unless they belong to a specific department.
• All tasks should have 2 retries by default.
• The default schedule for all dags should be daily at midnight (@daily), unless specified otherwise.

DAG Factory automatically looks for a file named defaults.yml in your dags folder and applies its configuration to all dags within that folder and its subfolders. This creates a single source of truth for your global defaults.

load_yaml_dags uses the same default path for both the configurations and the YAML files: the path set as dags_folder. You can override only the path where DAG Factory looks for configurations, by setting the defaults_config_path parameter.

To implement our company standards, create a new file at dags/defaults.yml with the following content:

schedule: "@daily"  # dag-specific arguments at root level

default_args:
  owner: "astronomer"
  retries: 2

The real power of this feature comes from inheritance. DAG Factory applies defaults.yml files hierarchically. A defaults.yml in a subfolder will inherit from its parent and can override any of the parent’s settings.

Let’s apply this to our scenario. We want to override the default owner for our Marketing and Finance departments, and also change the default schedule just for the Marketing department, to run dags at 1 AM rather than midnight for this department.

First, let’s create the folder structure:

airflow
└── dags
    ├── defaults.yml
    ├── marketing
    │   ├── defaults.yml
    │   └── marketing_dag.yml
    └── finance
        ├── defaults.yml
        └── finance_dag.yml

Now, create dags/marketing/defaults.yml to set a new schedule and owner:

schedule: "0 1 * * *"

default_args:
  owner: "astronomer-marketing"

And for the Finance department, create dags/finance/defaults.yml to override only the owner:

default_args:
  owner: "astronomer-finance"

Now that our defaults are in place, creating the actual dags is incredibly simple and clean.

Create dags/marketing/marketing_dag.yml:

marketing_dag:
  tasks:
    some_process:
      operator: airflow.providers.standard.operators.bash.BashOperator
      bash_command: "echo processing data"

And similarly, create dags/finance/finance_dag.yml:

finance_dag:
  tasks:
    some_process:
      operator: airflow.providers.standard.operators.bash.BashOperator
      bash_command: "echo processing data"

Notice how concise these definitions are. We don’t need to specify start_date, owner, or retries because they are all handled by our layered defaults.yml files. This allows you to write minimal dag configurations while maintaining centralized control over your project’s standards.

In the Airflow UI, you will see two new dags, each with a different set of inherited properties:

• marketing_dag: Inherits the schedule (0 1 * * *) and owner (astronomer-marketing) from its local defaults.yml, and retries from the global defaults.yml.
• finance_dag: Inherits the owner (astronomer-finance) from its local defaults.yml, and both the schedule (@daily) and retries from the global defaults.yml.

If any defaults.yml files are inside your dag_folder, DAG Factory might try to parse them as dags, which can cause errors in your task logs. To prevent this, keep dags_folder and defaults_config_path separate. Configuration inheritance still works as expected, and these errors are non-critical.

Advanced usage: Dynamic task mapping

DAG Factory also supports dynamic task mapping, to dynamically generate parallel tasks at runtime. The following example shows how to apply this principle using the TaskFlow API. Let’s assume we have the following Python functions defined in include/tasks/dtm_tasks.py:

def _generate_data():
    return [1, 2, 3, 4, 5]

def _process_data(processing_date, value):
    print(f"Processing {value} at {processing_date}")

We can now simply reference arguments under partial and expand in our YAML, to let DAG Factory apply dynamic task mapping:

dtm_example:
  default_args:
    owner: "astronomer"
    start_date: 2025-09-01
  schedule: "@hourly"

  tasks:
    generate_data:
      decorator: airflow.sdk.task
      python_callable: include.tasks.dtm_tasks._generate_data

    process_data:
      decorator: airflow.sdk.task
      python_callable: include.tasks.dtm_tasks._process_data
      partial:
        processing_date: "{{ logical_date }}"
      expand:
        value: +generate_data
      dependencies: [generate_data]

With this, we will use the output of generate_data to generate parallel task instances.

Advanced usage: Dynamic YAML generation

The examples above show how to use DAG Factory to create dags based on static YAML files. For use cases where you’d like to create several dags with a similar structure it is possible to create them dynamically based on a template YAML file to avoid code duplication.

Creating a dag dynamically with DAG Factory simply means that you use Python code to create the YAML configurations instead of writing them manually.

There are two files that you need:

• A template YAML file that contains the structure of the dags you want to create with placeholders for the values that will change.
• A Python script that creates DAG Factory YAML file by replacing the placeholders in the template YAML file with the actual values.

Since Airflow uses Jinja2 internally already, we can leverage this library for a more robust generation process.

The template YAML file provides the structure for all the dags you will generate dynamically with placeholders for values that vary in between the dags. Create a file called include/template.yml:

`include/template.yml`

{{ dag_id }}:
  schedule: "{{ schedule }}"
  tasks:
    task_1:
      operator: airflow.providers.standard.operators.bash.BashOperator
      bash_command: "{{ bash_command_task_1 }}"
    task_2:
      operator: airflow.providers.standard.operators.bash.BashOperator
      bash_command: "{{ bash_command_task_2 }}"
      dependencies: [task_1]

The Python script reads the template YAML file, replaces the placeholders with the actual values, and writes the resulting YAML files to the dags directory. Place this script in the top-level of your project for now. You can run this script manually to generate your dags for local development or automatically as part of your CI/CD pipeline.

`generate_yaml.py`

from pathlib import Path
import yaml
from jinja2 import Environment, FileSystemLoader

TEMPLATE_DIR = "include"
TEMPLATE_NAME = "template.yml"
OUTPUT_FILE = "dags/dynamic_dags.yml"
TEMPLATE_VARIABLES = [{
    "dag_id": "example_1",
    "schedule": "@daily",
    "bash_command_task_1": "echo task 1 from example 1",
    "bash_command_task_2": "echo task 2 from example 1",
}, {
    "dag_id": "example_2",
    "schedule": "@weekly",
    "bash_command_task_1": "echo task 1 from example 2",
    "bash_command_task_2": "echo task 2 from example 2",
}]

def generate_dags_from_template():
    # setup Jinja2
    env = Environment(loader=FileSystemLoader(TEMPLATE_DIR), autoescape=True)
    template = env.get_template(TEMPLATE_NAME)

    # render dags from template
    all_dags = {}
    for variables in TEMPLATE_VARIABLES:
        rendered_yaml_str = template.render(variables)
        dag_config = yaml.safe_load(rendered_yaml_str)
        all_dags.update(dag_config)

    # write to file
    output_path = Path(OUTPUT_FILE)
    with open(output_path, "w") as f:
        yaml.dump(all_dags, f, sort_keys=False)

    print(f"Successfully generated {len(TEMPLATE_VARIABLES)} dags into {OUTPUT_FILE}")

if __name__ == "__main__":
    generate_dags_from_template()

As a result, you will see the dynamically generated dags/dynamic_dags.yml file:

`dags/dynamic_dags.yml`

example_1:
  schedule: '@daily'
  tasks:
    task_1:
      operator: airflow.providers.standard.operators.bash.BashOperator
      bash_command: echo task 1 from example 1
    task_2:
      operator: airflow.providers.standard.operators.bash.BashOperator
      bash_command: echo task 2 from example 1
      dependencies:
      - task_1
example_2:
  schedule: '@weekly'
  tasks:
    task_1:
      operator: airflow.providers.standard.operators.bash.BashOperator
      bash_command: echo task 1 from example 2
    task_2:
      operator: airflow.providers.standard.operators.bash.BashOperator
      bash_command: echo task 2 from example 2
      dependencies:
      - task_1

Conclusion

In this tutorial, you’ve journeyed from defining a single dag in a YAML file to building a complete framework for dynamically generating your pipelines.

You’ve learned how to:

• Define dags, tasks, and task groups using a simple, declarative syntax.
• Pass data between tasks and use the TaskFlow API.
• Implement Airflow features like asset-aware scheduling and dynamic task mapping.
• Manage configuration at scale using hierarchical defaults.yml files for inheritance.
• Dynamically generate your YAML configurations using a templating engine.

Whether your goal is to empower analysts, standardize repetitive ETL jobs, or simply separate your pipeline’s structure from its logic, DAG Factory provides a robust, configuration-driven approach to Airflow development.

To continue your journey, explore the official DAG Factory repository, which contains many more examples and advanced use cases. You now have all the tools to start building your own dynamic dags.