Blueprint

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

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

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(BaseModel):
    model_config = ConfigDict(extra="forbid")
    source_table: str

DAG args config models are the opposite and need no such setting — Blueprint makes them strict for you (see Customizing DAG-Level Configuration).

Composing Templates

A blueprint can instantiate and render other blueprints inside its render() method, letting you build higher-level templates from lower-level building blocks while exposing a single, flat config to YAML authors.

Inside render(), instantiate each child blueprint, set its step_id, call render(...) with a config you construct, and wire the results together inside a parent TaskGroup:

class QualityGate(Blueprint[QualityGateConfig]):
    """Run checks then send a report — composed from Validate and Report."""

    def render(self, config: QualityGateConfig) -> TaskGroup:
        with TaskGroup(group_id=self.step_id) as group:
            validate = Validate()
            validate.step_id = "validate"
            validate_group = validate.render(ValidateConfig(checks=config.checks))

            report = Report()
            report.step_id = "report"
            report_task = report.render(ReportConfig(channel=config.report_channel))

            validate_group >> report_task
        return group

YAML authors then see a single step with a flat config, and the composed children stay invisible to them.

Composing DAGs in YAML

YAML Structure

# dags/my_pipeline.dag.yaml
dag_id: my_pipeline
schedule: "@daily"
description: "My data pipeline"

steps:
  step_one:
    blueprint: my_blueprint
    source_table: raw.customers
    batch_size: 500

  step_two:
    blueprint: another_blueprint
    depends_on: [step_one]
    target: analytics.output

By default, only schedule and description are supported as DAG-level fields (via the built-in DefaultDagArgs). For other fields like tags, default_args, catchup, etc., see Customizing DAG-Level Configuration.

Reserved Keys in Steps

Everything else passes to the blueprint's config.

Trigger Rules