blueprint

Blueprint 实现

你正在帮助用户使用 Blueprint。Blueprint 是一种基于 YAML 并利用可复用 Python 模板组合 Airflow DAG 的系统。按顺序执行步骤，并优先采用满足用户需求的最简配置。

软件包：PyPI 上的 airflow-blueprint — 本技能文档对应 0.5.0 仓库：https://github.com/astronomer/blueprint 要求：Python 3.10+、Airflow 2.5+ 交叉参考：使用 airflow 技能查找 Astro CLI、注册表和 REST API 的发现命令；如果用户需要完整的 Airflow 灵活性，而不是经过验证的模板，请使用 authoring-dags 或 dag-factory。

---

确定用户的需求

用户请求	操作
“创建蓝图” / “定义模板”	转到 创建蓝图
“从其他模板构建模板”	转到 组合模板
“从 YAML 创建 DAG” / “组合步骤”	转到 使用 YAML 组合 DAG
“在步骤或 DAG 之间复用值” / “每个环境使用不同值”	转到 变量和配置文件
“在现有 Python DAG 中使用蓝图” / “循环生成 DAG”	转到 在 Python DAG 中使用蓝图
“自定义 DAG 参数” / “为 DAG 添加标签” / “按文件夹设置不同的 DAG 默认值”	转到 自定义 DAG 级配置
“在不同仓库之间共享模板” / “从软件包安装蓝图”	转到 将蓝图作为软件包共享
“在运行时覆盖配置” / “使用参数触发”	转到 运行时参数覆盖
“对 DAG 进行后处理” / “添加回调” / “不要让一个错误文件导致一切失败”	转到 加载器选项
“验证我的 YAML” / “检查蓝图”	转到 验证命令
“在我的项目中设置蓝图”	转到 项目设置
“为蓝图设置版本”	转到 版本控制
“生成架构” / “设置 Astro IDE”	转到 架构生成
蓝图错误 / 故障排除	转到 故障排除

---

项目设置

如果用户从头开始，请指导他们完成设置：

1. 安装软件包

将 airflow-blueprint>=0.5.0 添加到 requirements.txt。

2. 创建加载器

创建 dags/loader.py：

from blueprint import build_all_airflow_dags

build_all_airflow_dags()

函数名称很重要。 Airflow 的安全模式 DAG 文件处理器只会解析同时包含 airflow 和 dag 的文件，因此，正是导入行本身让加载器能够被发现。build_all 和 build_all_dags 仍可作为已弃用的别名使用，但会发出 DeprecationWarning；请将现有加载器迁移到 build_all_airflow_dags。

DAG 级配置（调度计划、描述、标签、default_args 等）通过 YAML 字段和 BlueprintDagArgs 模板处理 — 请参阅 自定义 DAG 级配置。

3. 验证安装

在项目根目录运行 blueprint list。如果未找到蓝图，用户需要先创建蓝图类。

---

创建蓝图

标准示例

配置模型、泛型基类，以及一个返回以 self.step_id 为键的任务或组的 render()：请以此为基础进行调整，而不是另行设计不同的结构：

# 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

关键规则

元素	要求
配置类	必须继承自 BaseModel
蓝图类	必须继承自 Blueprint[ConfigClass]
render() 方法	必须返回 TaskGroup 或 BaseOperator
任务 ID	将 self.step_id 用作组 / 任务 ID
字段类型	必须是单一类型且兼容 YAML（见下文）

配置字段必须与 YAML 兼容

配置字段必须是单一类型。像 str | int 或 Union[A, B] 这样的多类型联合会在类定义时被拒绝（引发 TypeError），因为它们会导致 YAML 解析不明确，并生成 anyOf 架构。此检查会递归检查嵌套模型、列表项和字典值。

• 允许：标量（str、int、float、bool）、Literal[...]、list[X]、dict[str, V]、嵌套 BaseModel，以及 Optional[X] / X | None（可空模式）。
• 拒绝：str | int、Union[A, B]，或任何包含多个非 None 分支的联合类型。出于同样的原因，也会拒绝裸 Any 和 dict[str, Any]——请为值指定明确的单一类型。

无法通过 YAML 设置的内部字段

对于在 render() 内部使用且不应允许通过 YAML 覆盖的字段，请使用 Field(default=..., init=False)。这些字段会从构造函数中排除，也不会出现在 JSON Schema 输出中：

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

建议对步骤配置进行严格验证

步骤配置模型会继承 Pydantic 默认的 extra="ignore" 设置，因此步骤 YAML 中拼错的字段会被静默丢弃，而不会报告错误。建议使用 model_config = ConfigDict(extra="forbid")，将这些拼写错误转为错误：

class MyConfig(BaseModel):
    model_config = ConfigDict(extra="forbid")
    source_table: str

DAG 参数配置模型则相反，无需此类设置——Blueprint 会自动为你启用严格模式（参见自定义 DAG 级配置）。

---

组合模板

Blueprint 可以在其 render() 方法中实例化并渲染其他 Blueprint，让你能够使用较低层级的构建块构建更高层级的模板，同时向 YAML 编写者提供单一、扁平的配置。

在 render() 中，实例化每个子 Blueprint，设置其 step_id，使用你构造的配置调用 render(...)，并在父级 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 编写者看到的就是一个配置扁平的单一步骤，组合后的子项对他们不可见。

---

在 YAML 中组合 DAG

YAML 结构

# 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

默认情况下，只有 schedule 和 description 可用作 DAG 级字段（通过内置的 DefaultDagArgs）。对于 tags、default_args、catchup 等其他字段，请参见自定义 DAG 级配置。

步骤中的保留键

键	用途
blueprint	模板名称（必填）
depends_on	上游步骤名称列表
version	固定到特定 Blueprint 版本
trigger_rule	步骤的 Airflow 触发规则；根据已安装的 Airflow 版本进行验证

其他所有内容都会传递给 Blueprint 的配置。

触发规则

使用 trigger_rule 控制步骤相对于其上游依赖项的运行时机——例如，即使上游步骤失败，也运行通知步骤：

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

值会根据已安装的 Airflow 的 TriggerRule 枚举动态验证，因此可接受的取值取决于你的 Airflow 版本，而非此技能。当步骤的 Blueprint 渲染出 TaskGroup 时，该规则仅应用于组中的根任务（没有内部上游依赖项的任务），从而保留 Blueprint 编写者定义的内部连接关系。

Jinja2 支持

YAML 支持 Jinja2 模板，可访问环境变量、Airflow 变量/连接和运行时上下文：

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"

可用的模板变量：

• env — 环境变量
• var — Airflow Variables
• conn — Airflow Connections
• context — 用于生成 Airflow 模板表达式的代理，可访问运行时宏（例如 context.ds_nodash、context.dag_run.conf、context.task_instance.xcom_pull(...)）
• profile — 当前激活的变量配置名称；如果未选择配置，则为空。可用于根据配置推导值，而不必逐一列举每个配置：dag_id: "pipeline_{{ profile }}"

对于在解析时固定且由多个步骤或 DAG 共享的值，优先使用变量和配置，而不是 Jinja {% set %} 块——变量有作用域、可共享，并且对 blueprint lint 可见。

---

变量和配置文件

DAG YAML 可以声明变量，并以 ${name} 的形式引用它们。这样就无需在多个步骤和 DAG 中重复相同的值。

在 blueprint.vars.yaml 中声明变量，即可供其下的所有 DAG 共用；也可以在某个 DAG 自身的 vars: 块中声明，或两者兼用——就近声明会覆盖更上层的声明：

# 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}

变量替换在解析 YAML 后进行，因此 expiration_days 会保持为 int，而不是变成字符串 "90"。变量值可以是标量或列表，也可以组合使用（base: ${db}.${schema}）。

变量名必须匹配 ^[A-Za-z_][A-Za-z0-9_-]*$——允许使用连字符，并保留句点，以便日后可以无歧义地添加点分命名空间。

${...} 始终表示变量引用。 任何其他使用此语法的内容——最常见的是 bash_command 中的 shell 变量——都必须转义为 $${...}，否则 Blueprint 会尝试将其解析为变量。只有紧挨 { 前面的 $$ 会被视为转义，因此单独的 $$（shell PID、awk 字段）无需更改。blueprint lint 会报告每个未转义的用法，并在错误中说明转义方式，因此采用变量后请对项目运行 lint。

配置文件

变量可以针对不同的命名配置文件设置不同的值，并在构建时选择配置文件。环境是显而易见的用途，但此机制本质上只是按名称选择：

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

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

DAG 声明的每个配置文件都必须为该变量提供值；映射不完整会报错，而不会静默回退。

检查变量

blueprint vars <path> 会显示每个变量解析后的值及其来源，并标记 DAG 从未引用的变量。blueprint lint 会验证所有已声明的配置文件，除非使用 --profile 将范围缩小到某个配置文件。传入 --root，以匹配加载器构建时使用的路径；否则 lint 和运行时的解析结果会有所不同。

---

在 Python DAG 中使用 Blueprint

Blueprint 并不局限于 YAML 组合流程。以下两种模式可以在 Python 中使用 Blueprint，适用于逐步采用或根据数据生成 DAG。

在手写 DAG 中使用

实例化 Blueprint 类，设置其 step_id，调用 render()，并使用 >> 将其连接起来：

# 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

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"))

    setup >> extract_group

设置的 step_id 决定了 Blueprint 渲染时使用的 task_id / group_id。

使用 Builder / DAGConfig 以编程方式构建

若要根据数据生成 DAG（例如每个区域、租户等对应一个 DAG），可以在循环中构建 DAG，并将每个 DAG 注册到 globals() 中，以便 Airflow 发现它们：

from blueprint import Builder, DAGConfig

builder = Builder()

for region in ["us", "eu", "apac"]:
    config = DAGConfig(
        dag_id=f"pipeline_{region}",
        schedule="@hourly",
        steps={
            "extract": {"blueprint": "extract", "source_table": f"raw.{region}"},
        },
    )
    dag = builder.build(config, source_path=__file__)
    globals()[dag.dag_id] = dag

DAGConfig 接受与 YAML 中相同的字段。传入 source_path=__file__，即可从此文件所在目录解析 DAG 参数模板，方式与 YAML 文件相同；若不传入，则会回退到项目级默认值（参见 自定义 DAG 级配置）。

---

自定义 DAG 级配置

默认情况下，Blueprint 支持将 schedule 和 description 用作 DAG 级 YAML 字段。若要使用其他 DAG 构造函数参数（tags、default_args、catchup 等），请定义一个 BlueprintDagArgs 子类。其 render() 会返回一个字典，其中的 kwargs 将传递给 Airflow 的 DAG() 构造函数，因此可接受的键取决于你的 Airflow 版本中 DAG 所接受的参数。

# dags/dag_args.py
from typing import Any

from pydantic import BaseModel
from blueprint import BlueprintDagArgs

class ProjectDagArgsConfig(BaseModel):
    schedule: str | None = None
    tags: list[str] = []
    owner: str = "data-team"

class ProjectDagArgs(BlueprintDagArgs[ProjectDagArgsConfig]):
    def render(self, config: ProjectDagArgsConfig) -> dict[str, Any]:
        return {
            "schedule": config.schedule,
            "tags": config.tags,
            "default_args": {"owner": config.owner},
        }

这些已声明的字段随后会成为有效的 DAG 级 YAML 键，并由配置模型进行验证。

每个项目可有多个模板

一个项目可以定义多个模板。每个 DAG 使用其上方最近定义的模板：解析从 DAG 文件所在目录开始，沿父目录向上查找，因此子目录会覆盖其父目录中的模板。

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

如果 DAG 上方没有模板，则依次回退到声明了 default=True 的模板、唯一注册的模板，最后回退到内置的 DefaultDagArgs。模板的作用范围是定义它的 .py 文件所在目录，因此移动该文件会改变其作用范围——这是此功能中最常见的意外情况。

DAG YAML 无需任何更改：DAG 不会指定自己的模板。运行 blueprint list，查看各路径适用的模板、哪个模板是回退模板，以及每个模板的定义位置；blueprint lint 会列出每个 DAG 解析出的模板。

模板以其类名的 snake_case 形式注册——ProjectDagArgs 会变成 project_dag_args——这是 blueprint schema --dag-args <name> 所需的名称。设置 name = "..." 可以覆盖该名称，且名称本身必须采用 snake_case。

拒绝未声明的字段

DAG 参数配置模型定义了 DAG YAML 的顶层字段，因此 Blueprint 会自动对其应用 extra="forbid"——未声明的顶层键会报错，而不会被静默忽略。这与步骤配置的默认行为相反：除非选择启用，否则步骤配置会忽略未知键。

生成的架构中会显示 additionalProperties: false，因此编辑器和 Astro IDE 也会拒绝未知的顶层键。

如果你在模型上自行设置了 extra，则会保留你的设置。若要改为遵循模型自身的策略，可在类上设置 allow_extra=True：

class LooseDagArgs(BlueprintDagArgs[LooseConfig], allow_extra=True):
    ...

规则

• 同一目录中有两个模板会报错；两个模板同名（可通过在类上设置 name = "..." 来重命名其中一个）或有多个模板声明了 default=True，也会报错。
• 如果任何位置都没有子类，则使用内置的 DefaultDagArgs（仅包含 schedule 和 description）。

---

将 Blueprint 作为包共享

Blueprint 可以作为可安装的包在多个代码仓库间共享，而无需复制文件。该包通过 airflow_blueprint.blueprints 入口点组声明自身，安装后 Blueprint 会自动发现它，无需针对各代码仓库进行配置：

# pyproject.toml of the shared package
[project.entry-points."airflow_blueprint.blueprints"]
company_blueprints = "company_blueprints"

该值必须是普通的点分隔模块或包路径。声明的模块——如果是包，则包括其所有子模块——都会像本地发现的文件一样进行扫描，因此其中定义的任何 Blueprint 或 BlueprintDagArgs 子类都会被注册。

使用者安装该包后，Blueprint 会在 blueprint list 中与本地 Blueprint 一同显示，并通过来源列加以区分。导入失败的包会引发 EntryPointLoadError，而不会被静默地从注册表中移除。

若要关闭发现功能，可向加载器传入 discover_entry_points=False，或向 CLI 传入对应的 --no-entry-points 标志（使用 --help 确认哪些命令支持该选项）。

---

运行时参数覆盖

Blueprint 配置字段可在触发 DAG 时通过 Airflow 参数覆盖，让用户在手动触发 DAG 时自定义行为。

使用 supports_params = True 启用

Blueprint 必须将类属性 supports_params = True 设为启用，配置字段才会注册为 Airflow 参数（命名空间格式为 {step}__{field}）。**若未启用，self.param() / self.resolve_config() 不会产生任何效果，触发器表单中也不会显示任何字段。**仅对确实会使用这些方法的 Blueprint 启用，否则无用的参数会挤满表单，却不会产生任何效果。

标准示例

在运算符模板字段中使用 self.param()，由 Airflow 在执行时渲染其值；在 Python 可调用对象中使用 self.resolve_config()，以获取经过验证的配置对象。同一个 Blueprint 中可以同时使用这两种方法：

class Extract(Blueprint[ExtractConfig]):
    supports_params = True

    def render(self, config: ExtractConfig) -> TaskGroup:
        bp = self  # capture reference for the closure

        @task(task_id="run_query")
        def run_query(**context):
            resolved = bp.resolve_config(config, context)
            execute(resolved.query, resolved.batch_size)

        with TaskGroup(group_id=self.step_id) as group:
            BashOperator(
                task_id="shell_step",
                bash_command=f"run-etl --query {self.param('query')}",
            )
            run_query()
        return group

工作原理

• 参数会根据 Pydantic 配置模型自动生成，并按步骤命名空间隔离（例如 step_name__field）
• YAML 值会成为参数默认值；Pydantic 元数据（描述、约束、枚举值）会传递到 Airflow 触发器表单
• 无效的覆盖值会在执行时引发 ValidationError
• 可在触发器表单中覆盖这些值，或通过 DAG 运行端点提交带有命名空间名称的 conf（运行 af api ls --filter dagRun 可查找当前路径，另请参阅 airflow 技能）

自定义触发器表单

Pydantic 字段架构会传递到 Airflow 的触发器表单；json_schema_extra 控制每个字段的呈现方式（例如 format 中的多行文本和日期选择器、examples、values_display、description_md）。哪些设置会生效取决于 Airflow 版本，因此请检查表单，而不要想当然。

验证细节：只有可映射到 JSON Schema 的 Field 约束（ge、le、pattern、min_length、max_length、Literal 枚举）会在触发器表单中强制执行。自定义 @field_validator / @model_validator 逻辑无法映射到 JSON Schema，因此只会在构建时和 resolve_config() 内运行。如果自定义验证器负责强制执行重要约束，请在 @task 函数中调用 self.resolve_config()，使其对被覆盖的值进行验证。

---

加载器选项

build_all_airflow_dags() 接受用于控制整个项目的选项。以下选项会显著改变行为：

选项	作用
profile=	选择用于解析 ${...} 变量值的配置文件（参见 变量和配置文件）
skip_invalid_dags=True	渲染有效的 YAML 文件，并跳过有问题的文件，而不是让导入失败
discover_entry_points=False	关闭对以包形式安装的蓝图的发现
on_dag_built=	在每个 DAG 构建完成后进行后处理的回调

如果直接提供了 bp_registry，则会忽略 discover_entry_points，因为该注册表已经完成了发现。

签名的其余部分很少需要更改，主要用于内部流程：search_path 和 pattern 控制 YAML 发现，register_globals 会覆盖调用方的 globals()，render_templates 和 template_context 控制 Jinja，而 bp_registry 则提供预先构建的注册表。

使用 .airflowignore 排除文件

YAML 发现会遵循 .airflowignore，并使用 Airflow 自己的忽略文件遍历器，因此其语法、core.dag_ignore_file_syntax 设置和嵌套的忽略文件都与 DAG 处理器中的行为完全一致。在扫描目录时，blueprint lint 也会遵循该规则，因此被排除在 Airflow 之外的草稿也会被 lint 排除；显式传入该文件仍会对其执行 lint，可用于有意检查草稿。

匹配模式会针对每个路径的末尾部分，因此像 *.dag.yaml 这样的名称模式与使用 rglob 时的行为一致。在 Python 3.13 之前不支持 **。

构建后回调

使用 on_dag_built 在 DAG 构建完成后进行后处理，例如添加标签、访问控制或审计元数据：

from pathlib import Path
from blueprint import build_all_airflow_dags

def add_audit_tags(dag, yaml_path: Path) -> None:
    dag.tags.append("managed-by-blueprint")
    dag.tags.append(f"source:{yaml_path.name}")

build_all_airflow_dags(on_dag_built=add_audit_tags)

回调会接收已构建的 Airflow DAG（可变对象）以及定义该 DAG 的 YAML 文件的 Path。

跳过无效文件

skip_invalid_dags=True 可防止一个有问题的 YAML 文件导致文件夹中的其他 DAG 全部无法加载。推荐使用前请说明以下两点代价：

• 错误不再显示为 Airflow 导入错误，因为加载器本身可以正常解析。错误会进入 DAG 处理器日志，而这个位置远不如前者显眼。
• 重复的 DAG id 不再是错误：所有此类文件都会被解析，并以第一个 DAG 为准。

请在 CI 中配合使用 blueprint lint，以确保在显眼的位置发现无效文件。

---

验证命令

使用 uvx 运行 CLI 命令：

uvx --from airflow-blueprint blueprint <command>

命令	使用时机
list	显示可用的蓝图、版本、来源和 DAG 参数模板
describe <name>	显示蓝图的配置架构
lint	验证 DAG YAML——不带参数时递归扫描，或传入一个文件
vars <path>	显示 DAG 的解析后变量及每个变量的来源
schema	为蓝图或 DAG 级字段生成 JSON Schema
new	交互式创建 DAG YAML。--output-dir 用于指定文件的保存位置，该位置也决定了用于验证文件的 DAG 参数模板

每条命令都支持 --help，-h / -v 分别是 --help / --version 的简写。

请从项目根目录运行命令，而不要在 dags/ 目录内运行——不带参数的调用会相对于工作目录解析 dags/，因此在该目录内运行时找不到任何蓝图。其他目录布局请使用 --template-dir。

CLI 中的 Provider 运算符。 uvx --from airflow-blueprint 环境是隔离的，不包含 Astro Runtime 项目中的 Airflow Provider 包。如果模板导入了 Provider 运算符，请添加 --with <provider-package>，以便 CLI 导入这些运算符；否则 list/lint/schema 会因 ModuleNotFoundError 而失败：

uvx --from airflow-blueprint --with apache-airflow-providers-google blueprint list --template-dir dags/templates

---

版本管理

版本命名约定

版本是带有 V{N} 后缀的独立类：Extract 是 v1，ExtractV2 是 v2，并且每个类都有自己的配置模型。一个蓝图发现的版本必须构成连续的 1..N 序列。

class Extract(Blueprint[ExtractConfig]):        # v1
    def render(self, config): ...

class ExtractV2(Blueprint[ExtractV2Config]):    # v2, breaking changes
    def render(self, config): ...

显式名称和版本

当类名不符合约定时，可直接设置名称和版本：

class MyCustomExtractor(Blueprint[ExtractV3Config]):
    name = "extract"
    version = 3

    def render(self, config): ...

显式设置的 name 必须为 snake_case（^[a-z][a-z0-9_]*$），否则类会在定义时引发 ValueError。如果未设置，名称将采用类名对应的 snake_case 格式。

在 YAML 中使用版本

省略 version 即使用最新版本；固定指定版本可让某个步骤继续使用较旧版本：

steps:
  legacy_extract:
    blueprint: extract
    version: 1
    source_table: raw.data

blueprint list 会显示每个蓝图发现的版本。

---

架构生成

生成 JSON Schema，以支持编辑器自动补全或外部工具。blueprint schema <name> 会输出步骤模板的配置；blueprint schema --dag-args 会输出 DAG 级字段（dag_id、steps 以及 BlueprintDagArgs 提供的所有字段）。如果有多个 DAG 参数模板，--dag-args 可选地接受一个模板名称。

每个生成的架构都包含顶层 templateType 字段——步骤模板的值为 "blueprint"，DAG 级字段的值为 "dag_args"——以便使用者区分两者。通过管道传输或使用 -o/--output 写入时，命令会输出原始 JSON；交互式运行时，则会输出经过美化和语法高亮的 JSON。

请使用 -o/--output 写入，不要使用 >。 导入模板时可能会向 stdout 输出警告——常见情况是导入运算符时出现 Airflow 弃用警告——这些警告会与 JSON 交错，导致重定向后的输出无法解析。-o 只会写入架构。

可选字段会输出为普通类型。 可选配置字段会发布为 {"type": "string"}，而不是包含 null 分支的 anyOf——可选性仅通过架构的 required 数组表示。这样生成的客户端和表单渲染器就不会为每个可选字段都生成联合包装类型。Airflow 参数有意采用不同方式，保留可空类型，因为未设置的可选参数表示显式 null，而不是缺少键；不要为了让其中一种与另一种保持一致而“修正”它。

Astro 项目自动检测

创建或修改蓝图后，自动检查项目是否为 Astro 项目：查看是否存在 .astro/ 目录（该目录由 astro dev init 创建）。

如果是，自动重新生成架构，无需提示；根据 blueprint list 中的蓝图，为每个蓝图写入一个文件，并写入 DAG 级别的参数架构，保存到 blueprint/generated-schemas/。Astro IDE 会读取该目录以呈现配置表单，因此保持目录内容同步可确保可视化构建器反映最新配置。

如果无法判断项目是否为 Astro 项目，请询问用户一次，并在本次会话剩余时间内记住答案。

---

故障排查

错误消息中包含相应的修复提示；在采取此处的任何操作前，请先阅读错误消息。

“Blueprint not found”

原因：蓝图类不在 Python 路径中。

修复：使用 --template-dir 将 CLI 指向正确的目录，并通过 blueprint list 检查实际发现的内容。如果蓝图应来自已安装的软件包，请确认已启用入口点发现。

“Extra inputs are not permitted”

原因：启用 extra="forbid" 时，YAML 字段名称拼写错误。

修复：运行 blueprint describe <name> 查看有效字段名称。

DAG 未显示在 Airflow 中

原因：缺少加载器或加载器已损坏——包括导入已弃用别名、因而可能被 Airflow 安全模式跳过的加载器。

修复：确保 dags/loader.py 调用 build_all_airflow_dags()。如果设置了 skip_invalid_dags=True，即使 DAG 已损坏，文件也会解析成功；因此请检查 DAG 处理器日志，而不是导入错误视图。

CLI 出现 “ModuleNotFoundError: No module named 'airflow.providers.X'”

原因：独立的 uvx --from airflow-blueprint 环境不包含项目所用的 Airflow Provider 软件包，因此无法导入使用 Provider Operator 的模板。这是 CLI 的隔离环境问题，并非项目本身的问题。

修复：在 uvx 调用中添加 --with apache-airflow-providers-X。

未解析或意外的 ${...}

原因：某个 ${...} 不是已声明的变量——通常是 bash_command 中的 shell 变量，或是在搜索根目录之外的 blueprint.vars.yaml 中声明的变量。

修复：将非变量用法转义为 $${...}。对于确实缺失的变量，运行 blueprint vars <path> 查看解析结果，并运行 blueprint lint 查看完整列表；检查 --root 是否与加载器使用的路径一致。

“CyclicVariableError” / “CompositionDepthError”

原因：变量相互引用形成循环，或组合链深度超过解析器限制。

修复：错误消息会指出循环或链条。将其中一个值内联即可打破循环；blueprint vars <path> 会显示每个变量的解析结果。

“MultipleDagArgsError” / “DuplicateDagArgsError” / “MultipleDefaultDagArgsError”

原因：问题并非模板数量过多——多个模板是受支持的。只有在解析存在歧义时才会触发这些错误：同一目录中有两个模板、两个模板同名，或有多个模板声明了 default=True。

修复：将一个模板移到其 DAG 应使用的目录中，用 name = "..." 重命名其中一个，或只保留一个 default=True。blueprint list 会显示各处适用的模板。

“DagArgsNotFoundError”

原因：请求的具名 DAG 参数模板未注册。

修复：通过 blueprint list 检查名称。

“EntryPointLoadError”

原因：声明了蓝图的已安装软件包导入失败。

修复：直接导入模块以查看实际回溯信息，并确认软件包及其依赖项已安装在与 Airflow 相同的环境中。

“NonContiguousVersionError” / “InvalidVersionError”

原因：蓝图版本未按连续的 1..N 序列排列，或 YAML 固定了不存在的版本。

修复：确保版本逐一递增且没有缺漏；运行 blueprint list 查看可用版本。

“与 YAML 不兼容的字段”（导入时出现 TypeError）

原因：某个配置字段使用了 Blueprint 不接受的类型，例如多类型联合（如 str | int）、未指定类型的 Any 或 dict[str, Any]。

修复方法：使用单一、明确的类型。仍支持 Optional[X] / X | None。请参阅 创建 Blueprint → 配置字段类型必须与 YAML 兼容。

“检测到循环依赖”

原因：depends_on 引用形成循环。

修复方法：检查步骤依赖关系并移除循环。

在 Airflow UI 中调试

每个 Blueprint 任务在 渲染后的模板 中都有额外字段：

• blueprint_step_config — 解析后的 YAML 配置
• blueprint_step_code — Blueprint 的 Python 源代码

---

验证清单

完成前，请与用户确认：

• blueprint list 显示其模板，并且预期的 DAG args 模板会应用于每个路径
• blueprint lint 通过（不带参数时递归扫描所有 *.dag.yaml，也可以传入特定文件——传入目录路径会失败并提示 Is a directory）
• 如果使用了变量或配置文件，blueprint vars 会按预期解析
• dags/loader.py 存在且会调用 build_all_airflow_dags()
• DAG 出现在 Airflow UI 中，且没有解析错误

---

参考

• GitHub：https://github.com/astronomer/blueprint
• PyPI：https://pypi.org/project/airflow-blueprint/

Astro IDE

• Astro IDE Blueprint 文档：https://docs.astronomer.io/astro/ide-blueprint