/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, declaring shared variables or per-environment profiles, validating configurations, sharing templates as an installable package, 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 — this skill documents **0.5.0** > **Repo**: https://github.com/astronomer/blueprint > **Requires**: Python 3.10+, Airflow 2.5+ > **Cross-references**: the `airflow` skill for Astro CLI, registry, and REST API discovery commands; `authoring-dags` or `dag-factory` when the user needs full Airflow flexibility instead of validated templates.

---

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** | | "Reuse a value across steps or DAGs" / "Different value per environment" | Go to **Variables and Profiles** | | "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" / "Different DAG defaults per folder" | Go to **Customizing DAG-Level Configuration** | | "Share templates across repos" / "Install blueprints from a package" | Go to **Sharing Blueprints as a Package** | | "Override config at runtime" / "Trigger with params" | Go to **Runtime Parameter Overrides** | | "Post-process DAGs" / "Add callback" / "Don't let one bad file break everything" | Go to **Loader Options** | | "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 `airflow-blueprint>=0.5.0` to `requirements.txt`.

2. Create the Loader

Create `dags/loader.py`:

from blueprint import build_all_airflow_dags

build_all_airflow_dags()

> **The function name matters.** Airflow's safe-mode DAG file processor only parses files containing both `airflow` and `dag`, so the import line itself is what makes the loader discoverable. `build_all` and `build_all_dags` still work as deprecated aliases that emit `DeprecationWarning`; migrate existing loaders to `build_all_airflow_dags`.

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

Run `blueprint list` from the project root. If no blueprints are found, the user needs to create blueprint classes first.

---

Creating Blueprints

Canonical Example

Config model, generic base class, and a `render()` returning a task or group keyed on `self.step_id`. Adapt this rather than inventing a different 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):
    source_table: str = Field(description="Source table name")
    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

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 and omitted from JSON Schema output:

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

Recommend Strict Validation for Step Configs

A **step** config model inherits Pydantic's default `extra="ignore"`, so a misspelled field in a step's YAML is silently dropped rather than reported. Suggest `model_config = ConfigDict(extra="forbid")` to turn those typos into errors:

class MyConfig(