astronomer/agents/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

Use trigger_rule to control when a step runs relative to its upstream dependencies — for example, to run a notification step even if an upstream step failed:

steps:
  notify:
    blueprint: notify
    depends_on: [analyze]
    trigger_rule: all_done   # run regardless of whether analyze succeeded

Values are validated dynamically against the installed Airflow's TriggerRule enum, so the accepted set follows your Airflow version rather than this skill. When the step's blueprint renders a TaskGroup, the rule applies only to the group's root tasks (those with no internal upstream), preserving the blueprint author's internal wiring.

Jinja2 Support

YAML supports Jinja2 templating with access to environment variables, Airflow variables/connections, and runtime context:

dag_id: "{{ env.get('ENV', 'dev') }}_pipeline"
schedule: "{{ var.value.schedule | default('@daily') }}"
 
steps:
  extract:
    blueprint: extract
    output_path: "/data/{{ context.ds_nodash }}/output.csv"

Available template variables:

• env — environment variables
• var — Airflow Variables
• conn — Airflow Connections
• context — proxy that generates Airflow template expressions for runtime macros (e.g. context.ds_nodash, context.dag_run.conf, context.task_instance.xcom_pull(...))
• profile — the active variable profile name, or nothing when none is selected. Useful for deriving a value from the profile rather than enumerating it per profile: dag_id: "pipeline_{{ profile }}"

For values that are fixed at parse time and shared across steps or DAGs, prefer Variables and Profiles over a Jinja {% set %} block — variables are scoped, shareable, and visible to blueprint lint.

Variables and Profiles

DAG YAML can declare variables and reference them as ${name}. Use this to stop repeating a value across steps and DAGs.

Declare them in a blueprint.vars.yaml shared by every DAG beneath it, in a DAG's own vars: block, or both — nearer declarations override further ones:

# dags/blueprint.vars.yaml — shared by every DAG beneath it
vars:
  landing_dataset: raw_events
  warehouse_db: analytics

# dags/customer_etl.dag.yaml
vars:
  stream: customer_events
  retention_days: 90
 
steps:
  load:
    blueprint: load
    target_table: ${warehouse_db}.${landing_dataset}.${stream}
    expiration_days: ${retention_days}

Substitution runs after YAML parsing, so expiration_days stays an int rather than becoming the string "90". Values are scalars or lists, and variables may compose (base: ${db}.${schema}).

Variable names match ^[A-Za-z_][A-Za-z0-9_-]*$ — hyphens are allowed, and periods are reserved so dotted namespaces can be added later without ambiguity.

`${...}` is always a variable reference. Anything else that uses that syntax — most often a shell variable in a bash_command — must be escaped as $${...}, or Blueprint tries to resolve it as a variable. Only $$ immediately before { is treated as an escape, so a bare $$ (a shell PID, an awk field) needs no change. blueprint lint reports each unescaped occurrence and names the escape in the error, so lint the project after adopting variables.

Profiles

A variable can carry a different value per named profile, selected at build time. Environments are the obvious use, but the mechanism is just named selection:

profiles: [prod, dev]
vars:
  warehouse_db:
    prod: analytics
    dev: sandbox

build_all_airflow_dags(profile="prod" if is_production else "dev")

Every profile a DAG declares must give the variable a value; a partial mapping is an error rather than a silent fallback.

Inspecting Variables