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 enabli

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 : 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: 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: 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: 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 : 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 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 Key Purpose blueprint Template name (required) depends on List of upstream step names version Pin to specific blueprint version trigger rule Airflow trigger rule for the step; validated against the installed Airflow version 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: 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: 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: 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: Every profile a DAG declares must give the variable a value; a partial mapping is an error rather than a silent fallback. Inspecting Variables blueprint vars <path shows the resolved value of each variable and where it came from, and flags variables a DAG never references. blueprint lint validates every declared profile unless profile narrows it to one. Pass root to match the path the loader builds from, or resolution differs between lint and runtime. Blueprints in Python DAGs Blueprints aren't tied to the YAML composition flow. Two patterns let you use them from Python — useful for incremental adoption or data driven DAG generation. Inside a Hand Written DAG Instantiate the Blueprint class, set its step id , call render() , and wire it in with : The step id you set determines the task id / group id the blueprint renders under. Programmatic Building with Builder / DAGConfig For data driven DAG generation (one DAG per region, tenant, etc.), build DAGs in a loop and register each in globals() so Airflow discovers them: DAGConfig accepts the same fields you would write in YAML. Pass source path= file so the DAG args template is resolved from this file's directory the same way a YAML file's would be — without it, resolution falls back to the project wide default (see Customizing DAG Level Configuration ). Customizing DAG Level Configuration By default, Blueprint supports schedule and description as DAG level YAML fields. To use other DAG constructor arguments (tags, default args, catchup, etc.), define a BlueprintDagArgs subclass. Its render() returns a dict of kwargs passed to the Airflow DAG() constructor, so the accepted keys are whatever your Airflow version's DAG accepts. The declared fields then become valid DAG level YAML keys, validated by the config model. Several Templates per Project A project may define more than one template. Each DAG uses the template defined closest above it : resolution starts in the DAG file's own directory and walks up parent directories, so a subdirectory overrides its parents. A DAG with no template above it falls back to the one declared default=True , then to the sole registered template, then to the built in DefaultDagArgs . A template is scoped to the directory holding the .py file that defines it, so moving that file rescopes it — the most common surprise in this feature. Nothing in the DAG YAML changes: a DAG never names its template. Run blueprint list to see which template applies to which path, which one is the fallback, and where each is defined; blueprint lint names the resolved template per DAG. A template registers under the snake case form of its class name — ProjectDagArgs becomes project dag args — which is the name blueprint schema dag args <name expects. Setting name = "..." overrides it, and must itself be snake case. Undeclared Fields Are Rejected A DAG args config model defines the DAG YAML's top level surface, so Blueprint applies extra="forbid" to it automatically — an undeclared top level key is an error rather than a silently ignored one. This is the opposite default from step configs , which ignore unknown keys unless you opt in. This shows up in the generated schema as additionalProperties: false , so editors and the Astro IDE reject unknown top level keys too. Setting extra yourself on the model leaves your choice