/blueprint

name: blueprint
description: Define reusable Airflow task group templates with Pydantic validation and compose DAGs from YAML. Use when creating blueprint templates, composing DAGs from YAML, validating configurations, or enabling no-code DAG authoring for non-engineers.

Blueprint Implementation

You are helping a user work with Blueprint, a system for composing Airflow DAGs from YAML using reusable Python templates. Execute steps in order and prefer the simplest configuration that meets the user's needs.

> **Package**: `airflow-blueprint` on PyPI > **Repo**: https://github.com/astronomer/blueprint > **Requires**: Python 3.10+, Airflow 2.5+, Blueprint 0.3.0+

Before Starting

Confirm with the user: 1. **Airflow version** ≥2.5 2. **Python version** ≥3.10 3. **Use case**: Blueprint is for standardized, validated templates. If user needs full Airflow flexibility, suggest writing DAGs directly or using DAG Factory instead.

---

Determine What the User Needs

| User Request | Action | |--------------|--------| | "Create a blueprint" / "Define a template" | Go to **Creating Blueprints** | | "Build a template from other templates" | Go to **Composing Templates** | | "Create a DAG from YAML" / "Compose steps" | Go to **Composing DAGs in YAML** | | "Use a blueprint in an existing Python DAG" / "Generate DAGs in a loop" | Go to **Blueprints in Python DAGs** | | "Customize DAG args" / "Add tags to DAG" | Go to **Customizing DAG-Level Configuration** | | "Override config at runtime" / "Trigger with params" | Go to **Runtime Parameter Overrides** | | "Post-process DAGs" / "Add callback" | Go to **Post-Build Callbacks** | | "Validate my YAML" / "Lint blueprint" | Go to **Validation Commands** | | "Set up blueprint in my project" | Go to **Project Setup** | | "Version my blueprint" | Go to **Versioning** | | "Generate schema" / "Astro IDE setup" | Go to **Schema Generation** | | Blueprint errors / troubleshooting | Go to **Troubleshooting** |

---

Project Setup

If the user is starting fresh, guide them through setup:

1. Install the Package

# Add to requirements.txt
airflow-blueprint>=0.3.0

# Or install directly
pip install airflow-blueprint

2. Create the Loader

Create `dags/loader.py`:

from blueprint import build_all_dags

build_all_dags()

> **Use `build_all_dags`, not `build_all`.** The function was renamed in 0.3.0 so the loader's import line contains the substring `dag`, which Airflow's safe-mode DAG file processor requires — otherwise the file is silently skipped and no DAGs appear. `build_all` still works as a deprecated alias (emits `DeprecationWarning`); migrate existing loaders.

DAG-level configuration (schedule, description, tags, default_args, etc.) is handled via YAML fields and `BlueprintDagArgs` templates — see **Customizing DAG-Level Configuration**.

3. Verify Installation

uvx --from airflow-blueprint blueprint list

If no blueprints found, user needs to create blueprint classes first.

> **Provider operators in the CLI.** The `uvx --from airflow-blueprint` environment is isolated and does **not** include the Airflow provider packages your Astro Runtime project has. If your templates import provider operators (BigQuery, Snowflake, etc.), add `--with` so the CLI can import them — otherwise `list`/`lint`/`schema` fail with `ModuleNotFoundError: No module named 'airflow.providers.X'`: > > ```bash > uvx --from airflow-blueprint --with apache-airflow-providers-google blueprint list --template-dir dags/templates > ```

---

Creating Blueprints

When user wants to create a new blueprint template:

Blueprint Structure

# dags/templates/my_blueprints.py
from airflow.operators.bash import BashOperator
from airflow.utils.task_group import TaskGroup
from blueprint import Blueprint, BaseModel, Field

class MyConfig(BaseModel):
    # Required field with description (used in CLI output and JSON schema)
    source_table: str = Field(description="Source table name")
    # Optional field with default and validation
    batch_size: int = Field(default=1000, ge=1)

class MyBlueprint(Blueprint[MyConfig]):
    """Docstring becomes blueprint description."""

    def render(self, config: MyConfig) -> TaskGroup:
        with TaskGroup(group_id=self.step_id) as group:
            BashOperator(
                task_id="my_task",
                bash_command=f"echo '{config.source_table}'"
            )
        return group

Key Rules

| Element | Requirement | |---------|-------------| | Config class | Must inherit from `BaseModel` | | Blueprint class | Must inherit from `Blueprint[ConfigClass]` | | `render()` method | Must return `TaskGroup` or `BaseOperator` | | Task IDs | Use `self.step_id` for the group/task ID | | Field types | Must be single-typed and YAML-compatible (see below) |

Config Field Types Must Be YAML-Compatible

As of 0.3.0, config fields must be single-typed. Multi-type unions like `str | int` or `Union[A, B]` are **rejected at class-definition time** (raises `TypeError`) because they produce ambiguous YAML parsing and `anyOf` schemas. The check recurses through nested models, list items, and dict values.

• **Allowed**: scalars (`str`, `int`, `float`, `bool`), `Literal[...]`, `list[X]`, `dict[str, V]`, nested `BaseModel`, and `Optional[X]` / `X | None` (the nullable pattern).
• **Rejected**: `str | int`, `Union[A, B]`, or any union with more than one non-`None` arm. Bare `Any` and `dict[str, Any]` are rejected for the same reason — use an explicit single type for the value.

Internal Fields Not Settable from YAML

Use `Field(default=..., init=False)` for fields used inside `render()` that should not be overridable from YAML. They are excluded from the constructor (always use their default) and omitted from JSON Schema output:

class ExtractConfig(BaseModel):
    source_table: str
    _internal_batch_multiplier: int = Field(default=4, init=False)

Recommend Stric