blueprint

Blueprint の実装

再利用可能な Python テンプレートを使って YAML から Airflow DAG を構成するシステム、Blueprint をユーザーが利用できるよう支援します。手順は順番に実行し、ユーザーのニーズを満たす最もシンプルな設定を優先してください。

パッケージ: PyPI の airflow-blueprint — このスキルのドキュメントは 0.5.0 に対応しています リポジトリ: https://github.com/astronomer/blueprint 必要条件: Python 3.10 以降、Airflow 2.5 以降 相互参照: Astro CLI、レジストリ、REST API の検出コマンドについては airflow スキルを参照してください。ユーザーが検証済みテンプレートではなく、Airflow の柔軟性を十分に活用したい場合は authoring-dags または dag-factory を参照してください。

---

ユーザーのニーズを判断する

ユーザーのリクエスト	対応
「blueprint を作成する」／「テンプレートを定義する」	Blueprint の作成へ進みます
「他のテンプレートからテンプレートを構築する」	テンプレートの合成へ進みます
「YAML から DAG を作成する」／「ステップを合成する」	YAML での DAG の合成へ進みます
「ステップや DAG 間で値を再利用する」／「環境ごとに異なる値を使う」	変数とプロファイルへ進みます
「既存の Python DAG で blueprint を使う」／「ループで DAG を生成する」	Python DAG での Blueprint の使用へ進みます
「DAG の引数をカスタマイズする」／「DAG にタグを追加する」／「フォルダーごとに異なる DAG のデフォルトを設定する」	DAG レベルの設定のカスタマイズへ進みます
「リポジトリ間でテンプレートを共有する」／「パッケージから blueprint をインストールする」	パッケージとしての Blueprint の共有へ進みます
「実行時に設定をオーバーライドする」／「パラメーターを指定してトリガーする」	実行時のパラメーターのオーバーライドへ進みます
「DAG の後処理を行う」／「コールバックを追加する」／「不正なファイルがあってもすべてが停止しないようにする」	ローダーのオプションへ進みます
「YAML を検証する」／「blueprint をリントする」	検証コマンドへ進みます
「プロジェクトで blueprint をセットアップする」	プロジェクトのセットアップへ進みます
「blueprint をバージョン管理する」	バージョン管理へ進みます
「スキーマを生成する」／「Astro IDE をセットアップする」	スキーマの生成へ進みます
Blueprint のエラー／トラブルシューティング	トラブルシューティングへ進みます

---

プロジェクトのセットアップ

新規に始める場合は、次の手順でセットアップしてください。

1. パッケージをインストールする

requirements.txt に airflow-blueprint>=0.5.0 を追加します。

2. ローダーを作成する

dags/loader.py を作成します。

from blueprint import build_all_airflow_dags

build_all_airflow_dags()

関数名が重要です。 Airflow のセーフモード DAG ファイルプロセッサーは、airflow と dag の両方を含むファイルのみを解析するため、ローダーを検出可能にするのは import 行そのものです。build_all と build_all_dags は非推奨のエイリアスとして引き続き動作し、DeprecationWarning を出力します。既存のローダーは build_all_airflow_dags に移行してください。

DAG レベルの設定（スケジュール、説明、タグ、default_args など）は YAML フィールドと BlueprintDagArgs テンプレートで扱います。詳しくは DAG レベルの設定のカスタマイズを参照してください。

3. インストールを確認する

プロジェクトのルートから blueprint list を実行します。blueprint が見つからない場合は、まず blueprint クラスを作成する必要があります。

---

Blueprint の作成

標準的な例

設定モデル、汎用基底クラス、および 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 クラス	Blueprint[ConfigClass] を継承する必要があります
render() メソッド	TaskGroup または BaseOperator を返す必要があります
タスク ID	グループ／タスク ID には self.step_id を使用します
フィールドの型	単一の型で、YAML と互換性がある必要があります（以下を参照）

Config フィールドの型は YAML と互換性が必要です

Config フィールドは単一の型である必要があります。str | int や Union[A, B] のような複数型のユニオンは、あいまいな YAML 解析と anyOf スキーマを生成するため、クラス定義時に拒否されます（TypeError が発生します）。このチェックは、ネストされたモデル、リスト項目、辞書の値にも再帰的に適用されます。

• 許可される型：スカラー型（str、int、float、bool）、Literal[...]、list[X]、dict[str, V]、ネストされた BaseModel、および Optional[X] / X | None（nullable パターン）。
• 拒否される型：str | int、Union[A, B]、または None 以外の選択肢が複数あるユニオン。Any 単独と dict[str, Any] も同じ理由で拒否されます。値には明示的な単一の型を使用してください。

YAML から設定できない内部フィールド

YAML から上書きされないようにしつつ render() 内部で使用するフィールドには、Field(default=..., init=False) を使用します。これらはコンストラクターから除外され、JSON Schema の出力にも含まれません。

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

ステップ Config には厳格な検証を推奨します

ステップの Config モデルは、Pydantic のデフォルトである extra="ignore" を継承します。そのため、ステップの YAML にあるフィールド名の誤記は、エラーとして報告されずに黙って破棄されます。model_config = ConfigDict(extra="forbid") を指定すると、誤記をエラーとして扱えます。

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

DAG args の Config モデルはその逆で、この設定は必要ありません。Blueprint が厳格な検証を適用します（DAG レベルの設定をカスタマイズするを参照）。

---

テンプレートの組み合わせ

ブループリントは render() メソッド内で他のブループリントをインスタンス化してレンダリングできます。これにより、YAML の作成者には単一のフラットな Config を公開しながら、より低レベルの構成要素から高レベルのテンプレートを構築できます。

render() 内で子ブループリントをそれぞれインスタンス化し、step_id を設定して、構築した Config を指定して 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 の作成者からは単一のステップとフラットな Config だけが見え、組み合わせた子ブループリントは表示されません。

---

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

デフォルトでは、DAG レベルのフィールドとしてサポートされるのは schedule と description のみです（組み込みの DefaultDagArgs 経由）。tags、default_args、catchup などの他のフィールドについては、DAG レベルの設定をカスタマイズするを参照してください。

ステップで予約されているキー

キー	用途
blueprint	テンプレート名（必須）
depends_on	上流ステップ名のリスト
version	特定バージョンのブループリントに固定
trigger_rule	ステップの Airflow トリガールール。インストール済みの Airflow バージョンに対して検証されます

それ以外はすべてブループリントの Config に渡されます。

トリガールール

trigger_rule を使うと、上流の依存関係に対してステップを実行するタイミングを制御できます。たとえば、上流のステップが失敗しても通知ステップを実行できます。

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

値は、インストール済みの Airflow の TriggerRule enum に対して動的に検証されます。そのため、受け付けられる値はこのスキルではなく、使用している Airflow のバージョンに準拠します。ステップのブループリントが TaskGroup をレンダリングする場合、このルールはグループのルートタスク（内部に上流タスクがないタスク）にのみ適用され、ブループリントの作成者が設定した内部の接続は維持されます。

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間で同じ値を繰り返し記述せずに済みます。

すべての配下のDAGで共有されるblueprint.vars.yaml、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は文字列の"90"ではなく、intのままです。値にはスカラーまたはリストを指定でき、変数を組み合わせることもできます（base: ${db}.${schema}）。

変数名は^[A-Za-z_][A-Za-z0-9_-]*$に一致します。ハイフンは使用できますが、ドットは予約されています。将来、曖昧さなくドット区切りの名前空間を追加するためです。

${...}は常に変数参照として扱われます。 この構文を使うほかのもの（多くの場合、bash_command内のシェル変数）は、$${...}のようにエスケープする必要があります。そうしないと、Blueprintは変数として解決しようとします。{の直前にある$$だけがエスケープとして扱われるため、単独の$$（シェルの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で1つに絞らない限り、宣言されているすべてのプロファイルを検証します。ローダーがビルドに使うパスと一致させるには、--rootを指定してください。指定しないと、lint時と実行時で解決結果が異なります。

---

Python DAGでのBlueprint

BlueprintはYAMLの合成フローに限定されません。Pythonから使う方法は2つあります。段階的な導入や、データ駆動型の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生成（リージョンやテナントごとに1つの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で記述する場合と同じフィールドを指定できます。DAG引数テンプレートをYAMLファイルと同じ方法でこのファイルのディレクトリから解決するには、source_path=__file__を渡してください。これを指定しない場合、解決時にはプロジェクト全体のデフォルトが使われます（DAGレベルの設定をカスタマイズするを参照）。

---

DAGレベルの設定をカスタマイズする

デフォルトでは、BlueprintはDAGレベルのYAMLフィールドとしてscheduleとdescriptionをサポートしています。ほかのDAGコンストラクター引数（tags、default_args、catchupなど）を使うには、BlueprintDagArgsのサブクラスを定義します。そのrender()は、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 args の設定モデルは DAG YAML のトップレベル項目を定義するため、Blueprint は自動的に extra="forbid" を適用します。未宣言のトップレベルキーは、黙って無視されるのではなくエラーになります。これは、オプトインしない限り不明なキーを無視するステップ設定とは、デフォルトの動作が逆です。

生成されたスキーマでは additionalProperties: false として示されるため、エディターや Astro IDE でも不明なトップレベルキーは拒否されます。

モデルで extra を自分で設定した場合、その選択が維持されます。モデル独自のポリシーに委ねるには、クラスで allow_extra=True を指定します。

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

ルール

• 同じディレクトリにテンプレートが 2 つある場合、同じ名前のテンプレートが 2 つある場合（クラスに 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 にローカルのものと並んで表示され、ソース列で区別できます。パッケージのインポートに失敗すると、レジストリから黙って消えるのではなく EntryPointLoadError が発生します。

検出を無効にするには、ローダーに discover_entry_points=False を渡すか、CLI に対応する --no-entry-points フラグを指定します（どのコマンドが受け付けるかは --help で確認できます）。

---

実行時のパラメーター上書き

Blueprint の設定フィールドは、Airflow の params を使って DAG のトリガー時に上書きできるため、DAG を手動でトリガーする際にユーザーが動作をカスタマイズできます。

supports_params = True でオプトインする

設定フィールドを Airflow の params（{step}__{field} として名前空間化）として登録するには、Blueprint でクラス属性 supports_params = True を設定する必要があります。**これを設定しない場合、self.param() / self.resolve_config() は何もせず、トリガーフォームにもフィールドは表示されません。**これらのメソッドを実際に使う Blueprint のみオプトインしてください。使わない場合は、効果のない不要な params がフォームに表示されます。

基本的な例

Airflow が実行時に値をレンダリングするオペレーターのテンプレートフィールドでは self.param() を使います。検証済みの設定オブジェクトが必要な Python の呼び出し可能オブジェクトでは self.resolve_config() を使います。1 つの 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 を POST します（現在のパスは 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 の検出では Airflow 独自の除外ファイル走査機能を使い、.airflowignore が適用されます。そのため、構文、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（変更可能）と、それを定義した YAML ファイルの Path が渡されます。

無効なファイルをスキップする

skip_invalid_dags=True を指定すると、1 つの不正な YAML ファイルがフォルダー内のほかの DAG すべてに影響することを防げます。推奨する前に、次の両方のデメリットを説明してください。

• ローダー自体は正常に解析されるため、Airflow のインポートエラーとしては表面化しなくなります。代わりに DAG プロセッサーのログに出力されますが、ここははるかに見つけにくい場所です。
• DAG ID の重複がエラーではなくなります。該当するファイルはすべて解析され、最初の DAG が優先されます。

無効なファイルを見つけやすい場所で検出できるよう、CI で blueprint lint と組み合わせてください。

---

検証コマンド

CLI コマンドは uvx で実行します。

uvx --from airflow-blueprint blueprint <command>

コマンド	使用する場面
list	利用可能なブループリント、バージョン、ソース、DAG 引数テンプレートを表示します
describe <name>	ブループリントの設定スキーマを表示します
lint	DAG YAML を検証します。引数なしで再帰的にスキャンするか、ファイルを 1 つ指定します
vars <path>	DAG の解決済み変数と、それぞれの取得元を表示します
schema	ブループリントまたは DAG レベルのフィールドの JSON Schema を生成します
new	DAG YAML を対話形式で作成します。--output-dir でファイルの出力先を選択でき、検証に使用する DAG 引数テンプレートもそれに応じて選択されます

各コマンドは --help を受け付けます。また、-h / -v はそれぞれ --help / --version の短縮形として使えます。

dags/ の中ではなく、プロジェクトルートから実行してください。引数なしで実行すると、作業ディレクトリを基準に dags/ が解決されるため、その中から実行するとブループリントが見つかりません。それ以外のレイアウトでは --template-dir を使用してください。

CLI でのプロバイダーオペレーター。 uvx --from airflow-blueprint の環境は分離されており、Astro Runtime プロジェクトに含まれる Airflow プロバイダーパッケージは含まれていません。テンプレートがプロバイダーオペレーターをインポートする場合は、CLI からインポートできるように --with <provider-package> を追加してください。追加しないと、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 スキーマを生成します。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 を使って書き込んでください。 テンプレートのインポート時に、標準出力へ警告が出力されることがあります。よくある例は、オペレーターのインポートによる Airflow の非推奨警告です。こうした警告が JSON に混ざると、リダイレクトした出力を解析できなくなります。-o を使うとスキーマのみが書き込まれます。

オプションのフィールドは通常の型として出力されます。 オプションの設定フィールドは、null の分岐を含む anyOf ではなく、{"type": "string"} のように公開されます。オプションかどうかは、スキーマの required 配列だけで示されます。これにより、生成されたクライアントやフォームレンダラーが、オプションのフィールドごとにユニオンラッパー型を生成するのを防ぎます。Airflow のパラメーターは意図的に異なり、nullable 型を維持します。これは、未設定のオプションパラメーターがキーの不在ではなく、明示的な null を表すためです。一方をもう一方に合わせて「修正」しないでください。

Astro プロジェクトの自動検出

ブループリントを作成または変更した後、.astro/ ディレクトリ（astro dev init によって作成されます）があるかどうかを確認し、プロジェクトが Astro プロジェクトかどうかを自動的に確認します。

Astro プロジェクトの場合は、確認を求めずにスキーマを自動的に再生成します。blueprint list に表示される各ブループリント用のファイルと、DAG レベルの args スキーマを 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> を実行して、有効なフィールド名を確認してください。

Airflow に DAG が表示されない

原因：ローダーがないか、壊れています。非推奨のエイリアスをインポートするローダーも含まれます。この場合、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 プロバイダーパッケージが含まれていないため、プロバイダーのオペレーターをインポートするテンプレートを読み込めません。これは CLI の分離された環境であり、プロジェクトの環境ではありません。

対処方法：uvx の呼び出しに --with apache-airflow-providers-X を追加してください。

解決されない、または予期しない ${...}

原因：宣言されていない変数の ${...} です。よくある例として、bash_command 内のシェル変数や、検索ルートの外にある 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 args テンプレートが要求されました。

対処方法：名前が 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 タスクには、Rendered Template に追加フィールドがあります。

• blueprint_step_config — 解決済みの YAML 設定
• blueprint_step_code — Blueprint の Python ソース

---

検証チェックリスト

完了する前に、以下をユーザーと確認してください。

• blueprint list にテンプレートが表示され、想定される DAG 引数テンプレートが各パスに適用される
• blueprint lint が成功する（すべての *.dag.yaml を再帰的にスキャンする場合は引数なしで実行するか、特定のファイルを指定します。ディレクトリパスを指定すると Is a directory で失敗します）
• 変数またはプロファイルを使用している場合、blueprint vars が想定どおりに解決される
• dags/loader.py が存在し、build_all_airflow_dags() を呼び出している
• Airflow UI に DAG が解析エラーなしで表示される

---

参考資料

• 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