Blueprint

Reusable task group templates composed into Airflow DAGs via YAML.

What is Blueprint?

Blueprint lets data platform teams define reusable task group templates (Blueprints) in Python and compose them into Airflow DAGs using simple YAML files. Each Blueprint defines a validated Pydantic config and a render() method that produces a TaskGroup. DAGs are defined declaratively in YAML by composing blueprint instances as steps with explicit dependencies.

With Blueprint, you can:

• Define reusable task group templates with type-safe, validated configurations
• Compose DAGs from YAML by assembling blueprint instances as steps
• Version blueprints so DAGs can pin to specific template versions
• Get clear error messages when configs are invalid
• Use a CLI to list blueprints, validate YAML, and generate schemas
• See step config and blueprint source code in Airflow's rendered templates UI

Quick Start

1. Define Blueprint templates

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

class ExtractConfig(BaseModel):
    source_table: str = Field(description="Source table (schema.table)")
    batch_size: int = Field(default=1000, ge=1)

class Extract(Blueprint[ExtractConfig]):
    """Extract data from a source table."""

    def render(self, config: ExtractConfig) -> TaskGroup:
        with TaskGroup(group_id=self.step_id) as group:
            BashOperator(task_id="validate", bash_command=f"echo 'Validating {config.source_table}'")
            BashOperator(task_id="extract", bash_command=f"echo 'Extracting {config.batch_size} rows'")
        return group

class LoadConfig(BaseModel):
    target_table: str
    mode: str = Field(default="append", pattern="^(append|overwrite)$")

class Load(Blueprint[LoadConfig]):
    """Load data to a target table."""

    def render(self, config: LoadConfig) -> BashOperator:
        return BashOperator(
            task_id=self.step_id,
            bash_command=f"echo 'Loading to {config.target_table} ({config.mode})'"
        )

Blueprints typically return a TaskGroup containing multiple tasks. For simple cases, render() can also return a single BaseOperator -- the framework handles both uniformly.

2. Compose a DAG in YAML

# dags/customer_pipeline.dag.yaml
dag_id: customer_pipeline
schedule: "@daily"

steps:
  extract_customers:
    blueprint: extract
    source_table: raw.customers
    batch_size: 500

  extract_orders:
    blueprint: extract
    source_table: raw.orders

  load:
    blueprint: load
    depends_on: [extract_customers, extract_orders]
    target_table: analytics.customer_orders
    mode: overwrite

Step config is flat -- blueprint:, depends_on:, version:, and trigger_rule: are reserved keys; everything else is passed to the blueprint's config model. Steps with no depends_on run in parallel. Set trigger_rule: to control when a step runs relative to its upstream dependencies (e.g. one_success, all_done). The default is all_success. Valid values are determined by the installed Airflow version.

The blueprint: value is the snake_case form of the class name. Extract becomes extract, MultiSourceETL becomes multi_source_etl. See Template Versioning for details on how names and versions are determined.

3. Load DAGs

# dags/loader.py
from blueprint import build_all_airflow_dags

build_all_airflow_dags()

4. Validate

$ blueprint lint
PASS customer_pipeline.dag.yaml (dag_id=customer_pipeline)

Try It Out

The examples/ directory contains two working Airflow environments you can run locally using Docker and Tilt:

• Simple -- one DAG, two blueprints, zero advanced features
• Advanced -- space-themed, demonstrates many features

cd examples/simple/airflow3   # or airflow2
tilt up

See the examples README for full setup details.

DAG Arguments

By default, DAG YAML files support schedule and description at the top level (alongside dag_id and steps). For more control over DAG construction, define a BlueprintDagArgs template.

A BlueprintDagArgs subclass works like a Blueprint but for DAG-level arguments. It defines a Pydantic config model whose fields become the valid top-level YAML fields, and a render() method that returns DAG constructor kwargs.

from datetime import timedelta
from blueprint import BlueprintDagArgs, BaseModel, Field

class ProjectDagArgsConfig(BaseModel):
    schedule: str | None = None
    owner: str = "data-team"
    retries: int = Field(default=2, ge=0)

class ProjectDagArgs(BlueprintDagArgs[ProjectDagArgsConfig]):
    """Project-wide DAG argument template."""

    def render(self, config: ProjectDagArgsConfig) -> dict[str, Any]:
        kwargs: dict[str, Any] = {
            "default_args": {
                "owner": config.owner,
                "retries": config.retries,
            },
        }
        if config.schedule is not None:
            kwargs["schedule"] = config.schedule
        return kwargs

DAG YAML files then use the config fields directly:

dag_id: customer_pipeline
schedule: "@daily"
owner: analytics-team
retries: 3

steps:
  extract:
    blueprint: extract
    source_table: raw.customers

When no BlueprintDagArgs is defined, the built-in DefaultDagArgs provides schedule and description pass-through.

A project can define several templates. A DAG uses the one defined closest above it, so a subdirectory overrides its parents. The directory that scopes a template is the one holding the .py file that defines it:

dags/
  dag_args.py             ProjectDagArgs
  customer.dag.yaml       -> ProjectDagArgs
  sandbox/
    dag_args.py           SandboxDagArgs
    probe.dag.yaml        -> SandboxDagArgs

blueprint lint prints which template each DAG resolved to.

Two templates in one directory are ambiguous and raise an error, as do two sharing a name — set name = "..." on the class to register one under a different name. A DAG with no template above it uses the one declared as the fallback, or the only template defined:

class ProjectDagArgs(BlueprintDagArgs[ProjectDagArgsConfig], default=True):
    ...

DAGs built directly with the Builder API resolve from the Python file that builds them: pass source_path=__file__ to use the template above it.

Template Versioning

How names and versions are determined

By default, the blueprint name and version are inferred from the class name. The class name is converted to snake_case, and a trailing V{N} suffix is parsed as the version number:

Classes without a V{N} suffix are version 1.

Explicit name and version

When the class name doesn't match the desired blueprint name, set name and/or version as class attributes. This is useful when you want descriptive class names that don't dictate the registry identity:

class S3DataIngester(Blueprint[IngestConfig]):
    name = "ingest"
    version = 1

    def render(self, config: IngestConfig) -> TaskGroup: ...

class StreamingIngester(Blueprint[IngestV2Config]):
    name = "ingest"
    version = 2

    def render(self, config: IngestV2Config) -> TaskGroup: ...

Both register under ingest despite having unrelated class names. You can also set just one -- an explicit name with an inferred version from the class suffix, or an explicit version with a name inferred from the class name.

Versioning workflow

Each blueprint version is a separate class with its own config model. The initial version uses a clean name. Later versions add a V{N} suffix (or use explicit attributes). Breaking config changes are fine -- each version has an independent schema.

# v1 -- clean, no version thinking
class ExtractConfig(BaseModel):
    source_table: str
    batch_size: int = 1000

class Extract(Blueprint[ExtractConfig]):
    def render(self, config: ExtractConfig) -> TaskGroup:
        with TaskGroup(group_id=self.step_id) as group:
            BashOperator(task_id="validate", bash_command=f"echo 'Validating {config.source_table}'")
            BashOperator(task_id="extract", bash_command=f"echo 'Extracting {config.batch_size} rows'")
        return group

# v2 -- new class, new config, breaking changes are fine
class ExtractV2Config(BaseModel):
    sources: list[SourceDef]
    parallel: bool = True

class ExtractV2(Blueprint[ExtractV2Config]):
    def render(self, config: ExtractV2Config) -> TaskGroup:
        with TaskGroup(group_id=self.step_id) as group:
            for src in config.sources:
                BashOperator(task_id=f"extract_{src.table}", bash_command=f"echo 'Extracting {src.schema_name}.{src.table}'")
        return group

In YAML, pin to a version or omit to get the latest:

steps:
  # Pinned to v1
  extract_legacy:
    blueprint: extract
    version: 1
    source_table: raw.customers

  # Latest (v2)
  extract_new:
    blueprint: extract
    sources:
      - schema_name: raw
        table: orders

Sharing Blueprints Across Teams

In larger organizations, it's common that one "data engineering" team implements Blueprint templates and multiple other "dag authoring" teams leverage those templates. The data engineering team can publish those templates using a pip-installable package. It's a bad practice to copy-paste template code across repositories since that quickly goes out of sync.

Publishing Blueprint templates in a shared package requires declaring an entry point under the airflow_blueprint.blueprints group in your package's pyproject.toml:

# pyproject.toml
[project.entry-points."airflow_blueprint.blueprints"]
company_blueprints = "company_blueprints"

To leverage the Blueprint templates, install the package:

pip install company-blueprints  # or add it to requirements.txt

The Blueprint templates from the shared package become discoverable with the Blueprint CLI when the package is installed in your Python environment:

blueprint list

A few things worth knowing:

Collisions give an error

Two templates with the same name and version will raise a DuplicateBlueprintError. This also happens when templates are stored in different locations (such as locally and in package).

Keep the entry-point target scoped to code defining Blueprint templates

Every submodule under the entry-point gets scanned for Blueprint templates on every DAG parsing cycle. This could include unnecessary code. If your project contains other folders with non-Blueprint code, for example:

my_project/
├── pyproject.toml
└── my_project/
    ├── __init__.py
    ├── utils/       # Utility code, not Blueprint templates
    │   └── ...
    ├── operators/   # Custom Airflow operators, not Blueprint templates
    │   └── ...
    └── blueprints/  # <-- Only this contains Blueprint templates
        ├── __init__.py
        ├── extract.py
        └── load.py

A top-level entry-point will look like so:

# pyproject.toml
[project.entry-points."airflow_blueprint.blueprints"]
my_project = "my_project"

And limiting the entry-point to a subfolder is done like so:

# pyproject.toml
[project.entry-points."airflow_blueprint.blueprints"]
my_project = "my_project.blueprints"

Note that this applies to the discoverability of Blueprint templates. A template can still import from another module that's not included in the entry-point.

Airflow Rendered Templates

Every task instance gets two extra fields visible in Airflow's "Rendered Template" tab:

• blueprint_step_config -- the resolved YAML config for the step
• blueprint_step_code -- the full Python source file of the blueprint class

This makes it easy to understand what generated each task instance without leaving the Airflow UI.

Runtime Parameter Overrides

Blueprints that set supports_params = True have their config fields registered as Airflow DAG params, namespaced as {step}__{field}. When you trigger a DAG from the Airflow UI, the trigger form shows those fields pre-filled with YAML defaults — users can override any value before running.

Only blueprints that use self.param() or self.resolve_config() in their render() method should opt in — otherwise params would appear in the trigger form but have no effect.

Template access — self.param()

Returns a Jinja2 template string for use in operator template fields (e.g. bash_command, configuration). Airflow renders the actual value at execution time.

class Load(Blueprint[LoadConfig]):
    supports_params = True

    def render(self, config: LoadConfig) -> TaskGroup:
        with TaskGroup(group_id=self.step_id) as group:
            BashOperator(
                task_id="run_load",
                bash_command=f"echo 'Loading to {self.param('target_table')} mode={self.param('mode')}'",
            )
        return group

Variable access — self.resolve_config()

Merges runtime params into the Pydantic config inside a @task or PythonOperator callable. Returns a new validated config instance.

from airflow.decorators import task

class Load(Blueprint[LoadConfig]):
    supports_params = True

    def render(self, config: LoadConfig) -> TaskGroup:
        with TaskGroup(group_id=self.step_id) as group:
            @task(task_id="run_load")
            def run_load(**context):
                cfg = self.resolve_config(config, context)
                print(f"Loading to {cfg.target_table} ({cfg.mode})")
            run_load()
        return group

Both patterns can be combined in the same blueprint. Use self.param() for operators with template fields (BigQuery, CloudSQL, Bash, etc.) and self.resolve_config() for Python logic in @task functions.

Trigger form customization

Airflow's trigger form renders each param based on its JSON Schema. Blueprint passes schema metadata from your Pydantic fields through to Airflow, so you can control the form rendering using json_schema_extra:

class LoadConfig(BaseModel):
    query: str = Field(
        description="SQL to execute",
        json_schema_extra={"format": "multiline"},
    )
    schedule_date: str = Field(
        default="2024-01-01",
        json_schema_extra={"format": "date"},
    )

Supported format values include "multiline" (textarea), "date", "date-time", and "time" (pickers). You can also use examples (dropdown with free text), values_display (human-readable labels for enum/example values), and description_md (Markdown-formatted descriptions). See the Airflow params documentation for the full list of supported schema keys.

Validation behavior

Pydantic Field constraints that map to JSON Schema (ge, le, pattern, min_length, max_length, Literal enums) are enforced everywhere — at build time, in the Airflow trigger form, and in resolve_config().

Custom @field_validator and @model_validator logic does not map to JSON Schema and is therefore not enforced by the trigger form. These validators run at build time and in resolve_config() only.

If your config uses custom validators that enforce important constraints, use self.resolve_config() in @task functions to ensure those validators run on overridden values.

Complex config types

Scalar fields (str, int, float, bool) and Literal types render as native form controls in the trigger UI (text inputs, number inputs, dropdowns). Complex types work but render as JSON text inputs:

• Nested BaseModel fields → JSON object input
• Union types → JSON input with anyOf validation
• list[...] fields → JSON array input

For the best trigger form experience, prefer scalar fields for params that users will override frequently.

Triggering with overrides

Override params via the Airflow UI trigger form, or via the API using conf:

curl -X POST /api/v2/dags/customer_pipeline/dagRuns \
  -d '{"conf": {"load__target_table": "staging.customers", "load__mode": "append"}}'

Variables and Profiles

Declare variables once and reference them as ${name}. They resolve after YAML parsing, so a reference occupying an entire value keeps its type.

Project variables

Put shared values in blueprint.vars.yaml next to your DAGs:

# dags/blueprint.vars.yaml
vars:
  landing_dataset: raw_events
  warehouse_db: analytics

Using them in a DAG

A DAG can add its own variables, and override project ones:

dag_id: customer_etl

vars:
  stream: customer_events
  retention_days: 90

steps:
  load:
    blueprint: load
    target_table: ${warehouse_db}.${landing_dataset}.${stream}
    expiration_days: ${retention_days}

expiration_days stays an int because the whole value is a single reference. Embedding a reference in a larger string produces a string.

Profiles

Profiles are optional -- everything above works without declaring any. Declare them when a value must differ between environments, then key that variable by profile. Variables that do not vary stay plain values:

# dags/blueprint.vars.yaml
profiles: [prod, dev]

vars:
  landing_dataset: raw_events        # same everywhere

  warehouse_db:                      # differs per profile
    prod: analytics
    dev: sandbox

A DAG keys its own variables the same way, and may override only the profiles it needs to change:

dag_id: customer_etl
schedule: ${schedule}

vars:
  stream: customer_events
  schedule:
    prod: "@hourly"
    dev: "@daily"
  warehouse_db:                      # only dev differs; prod is inherited
    dev: team_sandbox

A profile-keyed variable must have a value for the profile being resolved, so a missing one fails rather than silently falling back to another profile's.

Selecting a profile

The loader picks which profile is active; every value stays in YAML:

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

blueprint lint                         # validates against every declared profile
blueprint lint --profile prod          # narrow to one
blueprint vars my.dag.yaml --profile prod --unused

A profile only has to be selected for variables a DAG actually references. A DAG using only invariant values needs no profile, even when the project declares them.

profiles: is declared once, in the outermost vars file; declaring it again in a nested file is an error.

Resolution order

Vars files are read from the search root down to the DAG's own directory, then its vars: block. Nearest wins:

blueprint.vars.yaml (search root) → blueprint.vars.yaml (subdirectory) → DAG `vars:` block

The search root is wherever build_all_airflow_dags() builds from; files above it are never read. blueprint lint and blueprint vars default their root to the current directory -- run them from the same place, or pass --root.

Values

A variable is a scalar or a list of scalars. A map is only ever a set of per-profile values — a map whose keys are not all declared profiles is an error rather than literal data, so ${a.b} never has a second possible meaning.

Group related values in the DAG rather than in a variable:

vars:
  bucket: s3://data
steps:
  load:
    paths:
      raw: ${bucket}/raw
      curated: ${bucket}/curated

Notes

• Variables may reference other variables (base: ${db}.${schema}); cycles are reported as errors.
• Variable names must start with a letter or underscore, followed by letters, digits, underscores or hyphens: [A-Za-z_][A-Za-z0-9_-]*.
• Write $${...} for a literal ${...} — a shell variable in a bash_command, for example. A bare $$ (the shell PID) is left alone.
• The active profile is available to Jinja2 as {{ profile }}, for the cases that genuinely need a conditional.

Variables or Jinja2?

Both work in a .dag.yaml, and Jinja2 renders first. Reach for ${...} for values that vary by environment: they are resolved before validation, keep their type, are checked by blueprint lint, and are safe for values containing YAML punctuation. Reach for {{ ... }} for Airflow runtime context ({{ context.ds }}), environment variables, and anything computed. Note that blueprint.vars.yaml is not Jinja2-rendered -- only .dag.yaml files are.

Jinja2 Templating in YAML

YAML files support Jinja2 templates with Airflow context:

dag_id: "{{ env.get('ENV', 'dev') }}_customer_etl"
schedule: "{{ var.value.etl_schedule | default('@daily') }}"

steps:
  extract:
    blueprint: extract
    source_table: "{{ var.value.source_schema }}.customers"

Airflow Runtime Context

Use the context accessor to pass through Airflow runtime macros that resolve at task execution time:

steps:
  extract:
    blueprint: extract
    date_partition: "{{ context.ds_nodash }}"
    output_path: "s3://bucket/{{ context.ds }}/data.parquet"

This renders at DAG parse time to literal Airflow template strings (e.g. {{ ds_nodash }}), which Airflow then resolves at task execution time. Chained access and function calls are supported:

steps:
  load:
    blueprint: load
    prev_result: "{{ context.ti.xcom_pull('extract') }}"

Note: Parse-time Jinja2 filters and arithmetic on context values are not supported. For complex expressions, use {% raw %}{{ ds | some_filter }}{% endraw %} instead.

Using Blueprints in Hand-Written Python DAGs

Blueprints aren't tied to the YAML composition flow. If you already have a regular Python DAG and just want to drop in a blueprint-rendered step amongst your existing tasks, you can instantiate the Blueprint class directly and call render() inside a with DAG(...) block:

# dags/hybrid_dag.py
from datetime import datetime

from airflow import DAG
from airflow.operators.bash import BashOperator

from dags.etl_blueprints import Extract, ExtractConfig, Load, LoadConfig


with DAG(
    dag_id="hybrid_python_dag",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
) as dag:
    setup = BashOperator(task_id="setup", bash_command="echo 'setup'")

    extract = Extract()
    extract.step_id = "extract"
    extract_group = extract.render(ExtractConfig(source_table="raw.events", batch_size=100))

    load = Load()
    load.step_id = "load"
    load_task = load.render(LoadConfig(target_table="warehouse.events", mode="append"))

    finalize = BashOperator(task_id="finalize", bash_command="echo 'done'")

    setup >> extract_group >> load_task >> finalize

Set step_id on the instance (it determines the task_id / group_id the blueprint renders under), then call render(config) to get back a BaseOperator or TaskGroup you can wire into the rest of your DAG. This is useful for incrementally adopting blueprints in an existing Python-DAG codebase without rewriting everything in YAML.

Programmatic Building

For advanced use cases, build DAGs programmatically:

from blueprint import Builder, DAGConfig

config = DAGConfig(
    dag_id="dynamic_pipeline",
    schedule="@hourly",
    steps={
        "step1": {"blueprint": "extract", "source_table": "raw.data"},
        "step2": {"blueprint": "load", "depends_on": ["step1"], "target_table": "out"},
    },
)

dag = Builder().build(config)

DAGConfig accepts the same fields you would write in YAML -- dag_id, steps, and any DAG-arg fields your BlueprintDagArgs consumes -- so this is handy when the set of DAGs is data-driven (one per region, tenant, or, in the example below, satellite). See examples/advanced/dags/programmatic_dags.py, which builds one DAG per satellite in a loop and registers each in globals() for Airflow to discover.

Post-Processing DAGs

The on_dag_built callback lets you modify each DAG after it's built from YAML. It receives the DAG and the path to the source YAML file:

# dags/loader.py
from pathlib import Path
from airflow import DAG
from blueprint import build_all_airflow_dags

def post_process(dag: DAG, yaml_path: Path) -> None:
    dag.tags = [*(dag.tags or []), "managed-by-blueprint"]
    dag.access_control = {"data-team": {"can_read", "can_edit"}}

build_all_airflow_dags(on_dag_built=post_process)

This is useful for applying cross-cutting concerns like access controls, tags, or custom metadata that shouldn't live in individual YAML files. The callback runs once per DAG, after all steps are wired up.

Ignoring DAG YAML Files

build_all_airflow_dags() honors .airflowignore files, with the same syntax and semantics as Airflow's DAG processor (including the core.dag_ignore_file_syntax setting and nested ignore files). A YAML file matched by an ignore entry is skipped:

# dags/.airflowignore
drafts

dags/
├── .airflowignore
├── loader.py
├── customer_pipeline.dag.yaml   # built
└── drafts/
    └── wip_pipeline.dag.yaml    # skipped

blueprint lint skips ignored files the same way when it discovers a directory tree. To lint an ignored file anyway, pass its path explicitly: blueprint lint drafts/wip_pipeline.dag.yaml.

Type Safety and Validation

Blueprint uses Pydantic for robust validation:

• Type coercion -- converts compatible types automatically
• Field validation -- min/max values, regex patterns, enums
• Custom validators -- add your own validation logic
• Clear error messages -- know exactly what went wrong

class ETLConfig(BaseModel):
    retries: int = Field(ge=0, le=5)
    timeout_minutes: int = Field(gt=0, le=1440)

    @field_validator('schedule')
    def validate_schedule(cls, v):
        valid = ['@once', '@hourly', '@daily', '@weekly', '@monthly']
        if v not in valid:
            raise ValueError(f'Must be one of {valid}')
        return v

Config Options for Template Authors

Pydantic offers model-level configuration that can make your Blueprint configs stricter or more flexible. Two options are particularly useful for YAML-based composition:

Rejecting Unknown Fields

By default, Pydantic silently ignores fields it doesn't recognize. This means a typo in a YAML step (e.g. batchsize instead of batch_size) is silently dropped and the default is used. Set extra="forbid" to catch this:

from pydantic import BaseModel, ConfigDict

class ExtractConfig(BaseModel):
    model_config = ConfigDict(extra="forbid")

    source_table: str
    batch_size: int = 1000

With this, batchsize: 500 in YAML raises a clear validation error instead of being silently ignored. This is recommended for configs where typos could cause hard-to-debug issues.

Internal Fields Not Settable from YAML

Template authors may want fields that exist on the config for use in render() but that cannot be overridden from YAML. Field(init=False) excludes a field from the constructor, so it always uses its default:

from pydantic import BaseModel, Field

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

init=False fields are excluded from the JSON Schema output since YAML authors cannot set them. This is useful for internal tuning parameters that should not be exposed as part of the public config interface.

Composing Templates

A blueprint can use other blueprints inside its render() method. This lets you build higher-level templates from lower-level building blocks while exposing a single, simplified config to YAML authors.

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


class ValidateConfig(BaseModel):
    checks: list[str] = Field(description="List of checks to run")

class Validate(Blueprint[ValidateConfig]):
    """Run data quality checks."""

    def render(self, config: ValidateConfig) -> TaskGroup:
        with TaskGroup(group_id=self.step_id) as group:
            for check in config.checks:
                BashOperator(task_id=check, bash_command=f"echo 'Running {check}'")
        return group


class ReportConfig(BaseModel):
    channel: str = Field(description="Notification channel")

class Report(Blueprint[ReportConfig]):
    """Send a quality report."""

    def render(self, config: ReportConfig) -> BashOperator:
        return BashOperator(
            task_id=self.step_id,
            bash_command=f"echo 'Sending report to {config.channel}'"
        )


class QualityGateConfig(BaseModel):
    checks: list[str] = Field(default=["nulls", "duplicates"])
    report_channel: str = Field(default="data-alerts")

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 see a single step with a flat config:

steps:
  quality:
    blueprint: quality_gate
    checks: [nulls, duplicates, freshness]
    report_channel: "#data-alerts"

Installation

uv add airflow-blueprint

CLI Commands

# List available blueprints
blueprint list

# Describe a blueprint's config schema
blueprint describe extract

# Describe a specific version
blueprint describe extract -v 1

# Validate DAG definitions (every declared profile, unless one is named)
blueprint lint pipeline.dag.yaml
blueprint lint --profile prod

# Show resolved variables and where each came from
blueprint vars pipeline.dag.yaml --profile prod --unused

# Generate JSON schema for editor support
# (each schema includes a top-level "templateType" field — "blueprint" for a
# step template, or "dag_args" for DAG-level fields via `blueprint schema --dag-args`)
blueprint schema extract > extract.schema.json

# A project with several DAG args templates has one DAG schema per template
blueprint schema --dag-args sandbox_dag_args > sandbox.dag.schema.json

# Create new DAG interactively
blueprint new

How is this different from DAG Factory?

DAG Factory exposes Airflow's full API via YAML. Blueprint hides that complexity behind safe, reusable task group templates with validation.

DAG Factory

my_dag:
  default_args:
    owner: 'data-team'
  schedule_interval: '@daily'
  tasks:
    extract_data:
      operator: airflow.operators.python.PythonOperator
      python_callable_name: extract_from_api
      python_callable_file: /opt/airflow/dags/etl/extract.py

Blueprint

dag_id: customer_pipeline
schedule: "@daily"

steps:
  extract:
    blueprint: extract
    source_table: raw.customers
  load:
    blueprint: load
    depends_on: [extract]
    target_table: analytics.customers

Use DAG Factory if: you need full Airflow flexibility and your users understand Airflow concepts.

Use Blueprint if: you want standardized, validated task group templates with type safety for teams.

Contributing

We welcome contributions! Please see our Contributing Guide for details.

License

Apache 2.0