This is an automated email from the ASF dual-hosted git repository.
uranusjr pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 337642574a6 YAML Dag file parser (#74084)
337642574a6 is described below
commit 337642574a653e54f111ff675adce0adc49ed918
Author: Tzu-ping Chung <[email protected]>
AuthorDate: Mon Oct 12 18:54:20 2026 +0800
YAML Dag file parser (#74084)
---
.pre-commit-config.yaml | 14 +
.../ci/prek/check_yaml_importer_schema_versions.py | 137 +++++
scripts/ci/prek/generate_yaml_importer_schema.py | 66 +++
task-sdk/pyproject.toml | 1 +
.../sdk/importers/yaml_importer/__init__.py | 32 ++
.../sdk/importers/yaml_importer/migrator.py | 168 ++++++
.../airflow/sdk/importers/yaml_importer/models.py | 459 ++++++++++++++++
.../airflow/sdk/importers/yaml_importer/parser.py | 102 ++++
.../sdk/importers/yaml_importer/schema.json | 365 +++++++++++++
.../importers/yaml_importer/versions/__init__.py | 38 ++
.../task_sdk/importers/yaml_importer/__init__.py | 16 +
.../importers/yaml_importer/example_dag.yaml | 69 +++
.../importers/yaml_importer/test_migrator.py | 188 +++++++
.../importers/yaml_importer/test_parser.py | 576 +++++++++++++++++++++
14 files changed, 2231 insertions(+)
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 3bbe42de850..3a2c3fb9661 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -1649,6 +1649,20 @@ repos:
pass_filenames: true
files:
^(task-sdk/src/airflow/sdk/execution_time/(comms\.py|schema/.*\.py)|airflow-core/src/airflow/dag_processing/processor\.py)$
require_serial: true
+ - id: generate-yaml-importer-schema-snapshot
+ name: Regenerate YAML DAG importer schema snapshot
+ entry: uv run -p 3.12 --frozen --no-progress --project task-sdk -s
scripts/ci/prek/generate_yaml_importer_schema.py
+ language: python
+ pass_filenames: false
+ files:
^task-sdk/src/airflow/sdk/importers/yaml_importer/(models|parser|versions/.*)\.py$
+ require_serial: true
+ - id: check-yaml-importer-schema-versions
+ name: Check YAML DAG importer schema changes have corresponding
version updates
+ entry: ./scripts/ci/prek/check_yaml_importer_schema_versions.py
+ language: python
+ pass_filenames: true
+ files:
^task-sdk/src/airflow/sdk/importers/yaml_importer/(models\.py|schema\.json|versions/.*\.py)$
+ require_serial: true
- id: generate-tasksdk-datamodels
name: Generate Datamodels for TaskSDK client
language: python
diff --git a/scripts/ci/prek/check_yaml_importer_schema_versions.py
b/scripts/ci/prek/check_yaml_importer_schema_versions.py
new file mode 100755
index 00000000000..d2190867485
--- /dev/null
+++ b/scripts/ci/prek/check_yaml_importer_schema_versions.py
@@ -0,0 +1,137 @@
+#!/usr/bin/env python
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# /// script
+# requires-python = ">=3.10,<3.11"
+# dependencies = [
+# "rich>=13.6.0",
+# ]
+# ///
+"""
+Fail when the YAML Dag schema changes without a matching version update.
+
+This mirrors :mod:`scripts.ci.prek.check_supervisor_schemas_versions` for the
+YAML Cadwyn bundle. Documents pin their format to a published ``$schema``, so a
+change to the generated snapshot that is not paired with a version update would
+silently reinterpret already-pinned documents once the current head date ships.
+
+The check is per-commit; if a schema-affecting source files or the committed
+snapshot changed, but nothing under ``versions/`` did, the hook asks the author
+to record the change as a new dated version. A ``versions/`` change is trusted.
+
+The comparison is against the committed ``schema.json`` on the target branch;
+the paired ``generate-yaml-importer-schema-snapshot`` hook guarantees the
+snapshot tracks the models, so the committed file is the source of truth here.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+from common_prek_utils import console, get_remote_for_main
+
+SCHEMA_DIR = "task-sdk/src/airflow/sdk/importers/yaml_importer/"
+SNAPSHOT_PATH = SCHEMA_DIR + "schema.json"
+MODELS_PATH = SCHEMA_DIR + "models.py"
+VERSIONS_PREFIX = SCHEMA_DIR + "versions/"
+
+
+def get_target_branch() -> str:
+ """Branch to compare against. GITHUB_BASE_REF for PRs, DEFAULT_BRANCH in
CI, else main."""
+ return os.environ.get("GITHUB_BASE_REF") or
os.environ.get("DEFAULT_BRANCH") or "main"
+
+
+def get_changed_files(filenames: list[str]) -> list[str]:
+ """Get changed files. Uses filenames from prek when provided, else staged
files for local runs."""
+ if filenames:
+ return filenames
+ result = subprocess.run(
+ ["git", "diff", "--cached", "--name-only"],
+ capture_output=True,
+ text=True,
+ check=True,
+ )
+ return [f for f in result.stdout.strip().splitlines() if f]
+
+
+def _target_ref() -> str:
+ """Fetch and return the target-branch ref to compare against."""
+ remote = get_remote_for_main()
+ target_branch = get_target_branch()
+ subprocess.run(["git", "fetch", remote, target_branch],
capture_output=True, check=False)
+ return f"{remote}/{target_branch}"
+
+
+def _snapshot_on_target(ref: str) -> str | None:
+ """Return the committed ``schema.json`` at *ref*, or ``None`` if it does
not exist there yet."""
+ result = subprocess.run(
+ ["git", "show", f"{ref}:{SNAPSHOT_PATH}"],
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+ return result.stdout if result.returncode == 0 else None
+
+
+def main() -> int:
+ changed_files = get_changed_files(sys.argv[1:])
+ schema_source_files = [f for f in changed_files if f in (MODELS_PATH,
SNAPSHOT_PATH)]
+ version_files = [f for f in changed_files if f.startswith(VERSIONS_PREFIX)]
+
+ if not schema_source_files:
+ return 0
+ if version_files:
+ # A version update accompanies the change -- trust the contributor.
+ return 0
+
+ ref = _target_ref()
+ target_snapshot = _snapshot_on_target(ref)
+ if target_snapshot is None:
+ # The snapshot is being introduced in this PR -- nothing on the target
+ # branch to compare against. The check activates once it merges.
+ console.print(
+ "[yellow]Skipping YAML importer schema version check:[/] target
branch "
+ "has no schema snapshot yet. The check activates once this PR
merges."
+ )
+ return 0
+
+ if Path(SNAPSHOT_PATH).read_text() != target_snapshot:
+ console.print("[bold red]ERROR:[/] YAML importer schema changed but no
version was updated.")
+ console.print("")
+ console.print("The following files were changed:")
+ for f in schema_source_files:
+ console.print(f" - [magenta]{f}[/]")
+ console.print("")
+ console.print(
+ f"The committed snapshot differs from [cyan]{ref}[/].\n"
+ "\n"
+ "Once a schema date is released, changing it reinterprets
documents pinned to that\n"
+ "version. Record the change as a new dated version carrying a
``VersionChange`` under:\n"
+ f" [cyan]{VERSIONS_PREFIX}[/]\n"
+ "(as the supervisor and execution-api bundles do), then regenerate
the snapshot."
+ )
+ return 1
+
+ console.print("[green]Snapshot unchanged:[/] source changes do not affect
the YAML importer schema.")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/scripts/ci/prek/generate_yaml_importer_schema.py
b/scripts/ci/prek/generate_yaml_importer_schema.py
new file mode 100755
index 00000000000..e785ea9826d
--- /dev/null
+++ b/scripts/ci/prek/generate_yaml_importer_schema.py
@@ -0,0 +1,66 @@
+#!/usr/bin/env python
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Regenerate the YAML Dag importer schema snapshot.
+
+The snapshot is the head-version JSON Schema generated from Pydantic models for
+the head ``$schema`` version (from the Cadwyn bundle). If the committed
snapshot
+at ``task-sdk/src/airflow/sdk/importers/yaml_importer/schema.json`` differs
from
+the freshly generated content, the hook rewrites it and exits non-zero.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+
+os.environ["_AIRFLOW__AS_LIBRARY"] = "1"
+
+from airflow.sdk.importers.yaml_importer.models import DagDocument
+from airflow.sdk.importers.yaml_importer.versions import get_bundle
+
+SNAPSHOT_PATH = (
+ Path(__file__)
+ .parents[3]
+ .joinpath("task-sdk", "src", "airflow", "sdk", "importers",
"yaml_importer", "schema.json")
+)
+
+
+def build_schema() -> str:
+ """Assemble the published head-version JSON Schema from the pydantic
models."""
+ head = get_bundle().versions[0].value # newest-first
+ schema = DagDocument.model_json_schema(by_alias=True)
+ schema["$schema"] = "https://json-schema.org/draft/2020-12/schema";
+ schema["title"] = f"Airflow YAML Dag ({head})"
+ return json.dumps(schema, indent=2) + "\n"
+
+
+def main() -> int:
+ new_content = build_schema()
+ if SNAPSHOT_PATH.exists() and SNAPSHOT_PATH.read_text() == new_content:
+ return 0
+ SNAPSHOT_PATH.write_text(new_content)
+ print(f"Regenerated {SNAPSHOT_PATH.name}. Please review the diff and
re-stage the file.")
+ return 1
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/task-sdk/pyproject.toml b/task-sdk/pyproject.toml
index 653d407f3d0..0ce16d4b713 100644
--- a/task-sdk/pyproject.toml
+++ b/task-sdk/pyproject.toml
@@ -52,6 +52,7 @@ dependencies = [
"asgiref>=3.11.1; python_version >= '3.14'",
"attrs>=24.2.0, !=25.2.0",
"babel>=2.17.0",
+ "cadwyn>=6.1.1",
"fsspec>=2023.10.0",
"httpx>=0.27.0",
"jinja2>=3.1.5",
diff --git a/task-sdk/src/airflow/sdk/importers/yaml_importer/__init__.py
b/task-sdk/src/airflow/sdk/importers/yaml_importer/__init__.py
new file mode 100644
index 00000000000..2f229838802
--- /dev/null
+++ b/task-sdk/src/airflow/sdk/importers/yaml_importer/__init__.py
@@ -0,0 +1,32 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Native YAML/JSON Dag format for Airflow."""
+
+from __future__ import annotations
+
+from airflow.sdk.importers.yaml_importer.models import DagDocument
+from airflow.sdk.importers.yaml_importer.parser import (
+ YamlDagParseError,
+ parse_documents,
+)
+
+__all__ = [
+ "DagDocument",
+ "YamlDagParseError",
+ "parse_documents",
+]
diff --git a/task-sdk/src/airflow/sdk/importers/yaml_importer/migrator.py
b/task-sdk/src/airflow/sdk/importers/yaml_importer/migrator.py
new file mode 100644
index 00000000000..9a27c07e4b9
--- /dev/null
+++ b/task-sdk/src/airflow/sdk/importers/yaml_importer/migrator.py
@@ -0,0 +1,168 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Schema version resolution and forward migration to the head shape.
+
+A document with an older ``$schema`` is migrated to the head shape by walking
+the Cadwyn version bundle and applying each ``VersionChange``'s forward request
+converters (``@convert_request_to_next_version_for(DagDocument)``), oldest to
+newest.
+
+Only whole-document request converters keyed on ``DagDocument`` are applied; a
+bundle carrying any other cadwyn instruction is rejected at construction rather
+than migrating a document with half its changes applied. (See documentation on
+:meth:`DagDocumentMigrator._reject_unsupported_instructions` for rationale.)
+"""
+
+from __future__ import annotations
+
+import copy
+import functools
+import re
+from typing import TYPE_CHECKING, Any
+
+import attrs
+
+from airflow.sdk.importers.yaml_importer.models import DagDocument
+from airflow.sdk.importers.yaml_importer.versions import get_bundle
+
+if TYPE_CHECKING:
+ from cadwyn import VersionBundle
+
+_SCHEMA_URL = "https://airflow.apache.org/schemas/dag/{date}.json";
+# The canonical ``$schema`` URL with the version captured; keep in sync with
``_SCHEMA_URL``.
+_SCHEMA_URL_RE =
re.compile(r"https://airflow\.apache\.org/schemas/dag/(\d{4}-\d{2}-\d{2})\.json")
+
+
+def schema_from_version(date: str) -> str:
+ """Build the canonical ``$schema`` URL for a published version *date*."""
+ return _SCHEMA_URL.format(date=date)
+
+
+def version_from_schema(schema: str) -> str | None:
+ """Read the version (``YYYY-MM-DD``) out of a canonical ``$schema`` URL,
or ``None``."""
+ match = _SCHEMA_URL_RE.fullmatch(schema)
+ return match.group(1) if match else None
+
+
[email protected]
+class _RequestInfo:
+ """
+ Duck-type stand-in for Cadwyn's ``RequestInfo``.
+
+ ``cadwyn.structure.data._AlterDataInstruction.__call__`` only reads
+ and writes ``info.body``; the by-schema transformers we drive never
+ touch FastAPI's Request/Response. Passing this minimal object lets
+ us run cadwyn's migrations from a pure in-process code path with no
+ HTTP stack.
+ """
+
+ body: dict[str, Any]
+
+
+def _calculate_version_values(migrator: DagDocumentMigrator) -> frozenset[str]:
+ return frozenset(v.value for v in migrator._bundle.versions)
+
+
[email protected]
+class DagDocumentMigrator:
+ """YAML Dag document migrator; pins each document to its ``$schema``
version."""
+
+ _bundle: VersionBundle
+ _version_values: frozenset[str] = attrs.field(
+ init=False,
+ default=attrs.Factory(_calculate_version_values, takes_self=True),
+ )
+
+ def __attrs_post_init__(self) -> None:
+ self._reject_unsupported_instructions()
+
+ def _reject_unsupported_instructions(self) -> None:
+ """
+ Fail if the bundle carries unsupported migrations.
+
+ Only forward request converters keyed on :class:`DagDocument` may run.
+ This is due to difficulties posed by things like templates and XCom,
+ which can present in nested locations inside task arguments.
Technically
+ we could just ban model-level migration only for them, but that's
+ considered unnecessarily complicated since you can already do all
+ per-model-level changes on :class:`DagDocument`. We can relax this
+ restriction in the future if the ergonomics prove to be suboptimal.
+ """
+ unsupported = (
+ "alter_endpoint_instructions",
+ "alter_enum_instructions",
+ "alter_request_by_path_instructions",
+ "alter_response_by_path_instructions",
+ "alter_response_by_schema_instructions",
+ "alter_schema_instructions",
+ )
+ for version in self._bundle.versions:
+ for change in version.changes:
+ name = change.__name__
+ other_models =
set(change.alter_request_by_schema_instructions) - {DagDocument}
+ if other_models:
+ raise RuntimeError(
+ f"{name}: request converters for {sorted(m.__name__
for m in other_models)} "
+ f"are not applied; only whole-document
{DagDocument.__name__} converters "
+ f"are supported"
+ )
+ if used := [b for b in unsupported if getattr(change, b,
None)]:
+ raise RuntimeError(
+ f"{name}: unsupported cadwyn instructions {used}; only
"
+ f"{DagDocument.__name__} request converters are
applied"
+ )
+
+ def resolve_version(self, version: str | None) -> str:
+ """Validate *version* is a published version in the bundle."""
+ if version is None or version not in self._version_values:
+ raise ValueError(f"$schema version {version!r} is not valid")
+ return version
+
+ def resolve_and_migrate(self, body: dict[str, Any]) -> dict[str, Any]:
+ """
+ Resolve *body*'s ``$schema`` version and migrate it to the head shape.
+
+ The version token is read from the ``$schema`` URL. Every valid
+ (published) version stays in the bundle and is migrated forward. An
+ unknown version is rejected.
+
+ :raises ValueError: if the ``$schema`` version is not valid.
+ :return: The migrated result. If *body* is already at head, it is
+ returned as-is; otherwise a migrated copy is returned.
+ """
+ if (schema_version := version_from_schema(raw_schema :=
body["$schema"])) is None:
+ raise ValueError(
+ f"$schema {raw_schema!r} is not a canonical schema URL "
+ f"(expected {_SCHEMA_URL.format(date='YYYY-MM-DD')!r})"
+ )
+ if (source_version := self.resolve_version(schema_version)) ==
self._bundle.versions[0].value:
+ return body
+ info = _RequestInfo(copy.deepcopy(dict(body)))
+ for version in self._bundle.reversed_versions:
+ if version.value <= source_version:
+ continue
+ for change in version.changes:
+ for instruction in
change.alter_request_by_schema_instructions.get(DagDocument, ()):
+ instruction(info) # type: ignore[arg-type]
+ return info.body
+
+
[email protected]
+def get_migrator() -> DagDocumentMigrator:
+ """Get the process-wide migrator bound to the format's version bundle."""
+ return DagDocumentMigrator(get_bundle())
diff --git a/task-sdk/src/airflow/sdk/importers/yaml_importer/models.py
b/task-sdk/src/airflow/sdk/importers/yaml_importer/models.py
new file mode 100644
index 00000000000..94fdd50f852
--- /dev/null
+++ b/task-sdk/src/airflow/sdk/importers/yaml_importer/models.py
@@ -0,0 +1,459 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Pydantic models for the YAML/JSON Dag format.
+
+These are the single source of truth for the format; they validate an authored
+document (rich Pydantic errors) and back the published JSON Schema. Only the
+format skeleton and value grammar are described; operator ``with:`` args,
+Dag- and task-level attributes pass through untyped because the Python
+constructors validate those at import time.
+"""
+
+from __future__ import annotations
+
+import collections
+import contextlib
+import datetime
+import itertools
+from typing import TYPE_CHECKING, Annotated, Any, ClassVar
+
+from pydantic import (
+ AliasChoices,
+ BaseModel,
+ ConfigDict,
+ Discriminator,
+ Field,
+ RootModel,
+ Tag,
+ model_validator,
+)
+
+if TYPE_CHECKING:
+ from pydantic import GetJsonSchemaHandler
+ from pydantic.json_schema import JsonSchemaValue
+ from pydantic_core import CoreSchema
+
+XCOM_KEYS = ("$x", "$xcom")
+TEMPLATE_KEYS = ("$t", "$template")
+CONST_KEY = "$const"
+
+# Keys deferred to a later edition...
+DAG_CALLBACK_KEYS = {"on_success_callback", "on_failure_callback"}
+TASK_CALLBACK_KEYS = {
+ "on_success_callback",
+ "on_failure_callback",
+ "on_retry_callback",
+ "on_execute_callback",
+ "on_skipped_callback",
+}
+TASK_ASSET_IO_KEYS = {"inlets", "outlets"}
+
+
+def _as_exclusive_union(schema: JsonSchemaValue) -> JsonSchemaValue:
+ """
+ Rewrite a disjoint ``anyOf`` union in place to ``oneOf``.
+
+ Pydantic emits ``anyOf`` (one or more match) for ``X | Y`` unions, but when
+ the branches are mutually exclusive, ``oneOf`` (exactly one match) accepts
+ the same values but generates better client code.
+
+ Only call this on unions whose branches are provably disjoint; on an
+ overlapping union, ``oneOf`` incorrectly rejects valid values.
+
+ This operation is performed in-place. The same modified object is returned.
+ """
+ with contextlib.suppress(KeyError):
+ schema["oneOf"] = schema.pop("anyOf")
+ return schema
+
+
+class XComTarget(BaseModel):
+ """The object form of an XCom reference."""
+
+ task: str
+ key: str | None = None
+
+ model_config = ConfigDict(extra="forbid")
+
+ @classmethod
+ def __get_pydantic_json_schema__(
+ cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler
+ ) -> JsonSchemaValue:
+ schema = handler(core_schema)
+ _as_exclusive_union(schema["properties"]["key"])
+ return schema
+
+
+class _MarkerRef(BaseModel):
+ """Base for a reserved ``$``-marker object."""
+
+ model_config = ConfigDict(extra="forbid")
+
+ _marker_keys: ClassVar[tuple[str, ...]]
+
+ @classmethod
+ def __get_pydantic_json_schema__(
+ cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler
+ ) -> JsonSchemaValue:
+ schema = handler(core_schema)
+ value_schema = _as_exclusive_union(next(iter(schema.get("properties",
{}).values())))
+ schema["properties"] = {key: value_schema for key in cls._marker_keys}
+ schema["additionalProperties"] = False
+ if len(cls._marker_keys) > 1: # Pydantic sets the correct 'required'
for the single-variant case.
+ schema.pop("required", None)
+ schema["oneOf"] = [{"required": [key]} for key in cls._marker_keys]
+ return schema
+
+
+class XComRef(_MarkerRef):
+ """An upstream task's XCom output."""
+
+ _marker_keys = XCOM_KEYS
+
+ target: str | XComTarget = Field(
+ serialization_alias="$x",
+ validation_alias=AliasChoices(*XCOM_KEYS),
+ )
+
+
+class TemplateRef(_MarkerRef):
+ """A Jinja template."""
+
+ _marker_keys = TEMPLATE_KEYS
+
+ source: str = Field(serialization_alias="$t",
validation_alias=AliasChoices(*TEMPLATE_KEYS))
+
+
+class ConstRef(_MarkerRef):
+ """Force a literal; the value is taken verbatim."""
+
+ _marker_keys = (CONST_KEY,)
+
+ value: Any = Field(serialization_alias="$const",
validation_alias=AliasChoices(CONST_KEY))
+
+
+def _value_discriminator(v: Any) -> str:
+ """
+ Route a raw value to a marker branch, or to ``literal``.
+
+ The prefix ``$`` is reserved to express Airflow constructs. To prevent
+ typos, any unknown ``$``-prefixed strings are routed here and rejected.
+
+ If such a key is needed, it can be wrapped in ``$const``, which would
+ avoid this discriminator (only used by :class:`Literal`).
+ """
+ if isinstance(v, dict) and len(v) == 1:
+ (key,) = v
+ if key in XCOM_KEYS:
+ return "xcom"
+ if key in TEMPLATE_KEYS:
+ return "template"
+ if key == CONST_KEY:
+ return "const"
+ if isinstance(v, XComRef):
+ return "xcom"
+ if isinstance(v, TemplateRef):
+ return "template"
+ if isinstance(v, ConstRef):
+ return "const"
+ return "literal"
+
+
+class Literal(RootModel):
+ """
+ A literal value.
+
+ Containers are validated recursively so nested markers resolve; scalars
pass
+ through. A ``$``-prefixed key here is a misused marker, so it is rejected.
+ """
+
+ root: dict[str, Value] | list[Value] | str | int | float | bool |
datetime.datetime | datetime.date | None
+
+ model_config = ConfigDict(
+ json_schema_extra={
+ "not": {"type": "object", "not": {"patternProperties": {r"^\$":
False}}},
+ "description": "A literal value.",
+ }
+ )
+
+ @model_validator(mode="before")
+ @classmethod
+ def _reject_reserved_keys(cls, v: Any) -> Any:
+ if isinstance(v, dict):
+ if (res := next((k for k in v if isinstance(k, str) and
k.startswith("$")), None)) is not None:
+ raise ValueError(
+ f"unexpected key {res!r}: '$'-prefixed keys are reserved "
+ f"markers; wrap a literal '$' key in $const"
+ )
+ return v
+
+ @classmethod
+ def __get_pydantic_json_schema__(
+ cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler
+ ) -> JsonSchemaValue:
+ # Pydantic generates a pretty messy schema from this. Clean it up since
+ # essentially we accept "whatever the format can implicitly accept".
+ schema = handler(core_schema)
+ schema["oneOf"] = [
+ *(b for b in schema.pop("anyOf") if b.get("type") in ("object",
"array")),
+ {"type": "boolean"},
+ {"type": "null"},
+ {"type": "number"},
+ {"type": "string"},
+ ]
+ return schema
+
+
+Value = Annotated[
+ Annotated[XComRef, Tag("xcom")]
+ | Annotated[TemplateRef, Tag("template")]
+ | Annotated[ConstRef, Tag("const")]
+ | Annotated[Literal, Tag("literal")],
+ Discriminator(_value_discriminator),
+]
+
+Literal.model_rebuild()
+
+
+class Task(BaseModel):
+ """
+ A ready-made operator (``uses``) or task with custom code (``run``).
+
+ At most one of ``uses`` or ``run`` may carry a body; ``null`` or omission
+ means no body. A task with neither will be turned by the importer into an
+ ``EmptyOperator``. Unknown keys pass through untyped as operator arguments.
+ """
+
+ id_: str = Field(alias="id", min_length=1)
+ needs: list[str] = Field(default_factory=list)
+ extends: list[str] = Field(default_factory=list)
+ with_: dict[str, Value] = Field(default_factory=dict, alias="with")
+ uses: str | None = Field(default=None, min_length=1)
+ run: dict[str, Value] | None = None
+
+ model_config = ConfigDict(
+ extra="allow",
+ json_schema_extra={"description": "A ready-made operator (uses) or
task with custom code (run)."},
+ )
+
+ @model_validator(mode="before")
+ @classmethod
+ def _check_task(cls, data: Any) -> Any:
+ if isinstance(data, dict):
+ for k in TASK_ASSET_IO_KEYS:
+ if k in data:
+ raise ValueError(f"{k!r} not yet implemented")
+ for k in TASK_CALLBACK_KEYS:
+ if k in data:
+ raise ValueError(f"{k!r} (callback) not yet implemented")
+ if sum(1 for k in ("uses", "run") if data.get(k) is not None) > 1:
+ raise ValueError(f"task {data.get('id')!r} may not set both
'uses' and 'run'")
+ return data
+
+ @classmethod
+ def __get_pydantic_json_schema__(
+ cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler
+ ) -> JsonSchemaValue:
+ schema = handler(core_schema)
+ value_schema = schema["properties"]["with"]["additionalProperties"]
+ schema["properties"]["uses"] = {"type": ["string", "null"],
"minLength": 1}
+ schema["properties"]["run"] = {"type": ["object", "null"],
"additionalProperties": value_schema}
+ schema["properties"]["extends"] = {
+ "oneOf": [{"type": "string"}, {"type": "array", "items": {"type":
"string"}}]
+ }
+ schema["not"] = {
+ "allOf": [
+ {"required": ["uses"], "properties": {"uses": {"type":
"string"}}},
+ {"required": ["run"], "properties": {"run": {"type":
"object"}}},
+ ]
+ }
+ return schema
+
+
+class TaskTemplate(BaseModel):
+ """
+ A reusable, partial task fragment merged into any task that ``extends`` it.
+
+ This holds arbitrary keys as-is. All validation is done on the merged task.
+ """
+
+ model_config = ConfigDict(
+ extra="allow",
+ json_schema_extra={
+ "description": (
+ "A reusable, partial task fragment merged into any task that "
+ "extends it.\n\nThis holds arbitrary keys as-is. All "
+ "validation is done on the merged task."
+ ),
+ },
+ )
+
+
+class TimetableSchedule(BaseModel):
+ """A constructed timetable."""
+
+ uses: str = Field(min_length=1)
+ with_: dict[str, Any] = Field(default_factory=dict, alias="with")
+
+ model_config = ConfigDict(extra="forbid")
+
+
+Schedule = str | None | TimetableSchedule
+
+
+def _merge_task(base: dict, over: dict) -> dict:
+ """
+ Shallow merge *over* onto *base*.
+
+ Note that ``with`` and ``run`` need to merge one level deep. All other
+ fields simply replace.
+ """
+ out = base.copy()
+ for k, v in over.items():
+ if k in ("with", "run") and isinstance(v, dict) and
isinstance(out.get(k), dict):
+ merged = dict(out[k])
+ merged.update(v)
+ out[k] = merged
+ else:
+ out[k] = v
+ return out
+
+
+def _validate_extends(value: Any, *, where: str) -> list[str]:
+ """
+ Return *value* as the list of template names it must be, or raise a clear
``ValueError``.
+
+ A bare string is accepted as shorthand for a single-element list.
+ """
+ if value is None:
+ return []
+ if isinstance(value, str):
+ return [value]
+ if not isinstance(value, list) or not all(isinstance(name, str) for name
in value):
+ raise ValueError(f"{where}: 'extends' must be a list of template
names, got {value!r}")
+ return value
+
+
+def _expand_template(name: str, templates: dict[str, Any], _seen: tuple = ())
-> dict:
+ if name in _seen:
+ raise ValueError(f"template cycle via {name!r}")
+ try:
+ spec = templates[name]
+ except KeyError:
+ raise ValueError(f"unknown template {name!r}") from None
+ if not isinstance(spec, dict):
+ raise ValueError(f"template {name!r} must be a mapping")
+ acc: dict = {}
+ for parent in _validate_extends(spec.get("extends"), where=f"template
{name!r}"):
+ acc = _merge_task(acc, _expand_template(parent, templates, _seen +
(name,)))
+ own = {k: v for k, v in spec.items() if k != "extends"}
+ return _merge_task(acc, own)
+
+
+def _iter_xcom_task_ids(value: Any):
+ """Yield task ids referenced by any XComRef inside a resolved value."""
+ if isinstance(value, XComRef):
+ yield value.target if isinstance(value.target, str) else
value.target.task
+ elif isinstance(value, Literal):
+ yield from _iter_xcom_task_ids(value.root)
+ elif isinstance(value, dict):
+ for v in value.values():
+ yield from _iter_xcom_task_ids(v)
+ elif isinstance(value, list):
+ for v in value:
+ yield from _iter_xcom_task_ids(v)
+
+
+class DagDocument(BaseModel):
+ """
+ A Dag represented by one YAML document.
+
+ Extra top-level keys are Dag-level arguments.
+ """
+
+ schema_: str = Field(alias="$schema")
+ dag_id: str = Field(min_length=1)
+ schedule: Schedule = None
+ templates: dict[str, TaskTemplate] = Field(default_factory=dict)
+ tasks: list[Task] = Field(default_factory=list)
+
+ model_config = ConfigDict(extra="allow")
+
+ @classmethod
+ def __get_pydantic_json_schema__(
+ cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler
+ ) -> JsonSchemaValue:
+ schema = handler(core_schema)
+ _as_exclusive_union(schema["properties"]["schedule"])
+ return schema
+
+ @model_validator(mode="before")
+ @classmethod
+ def _reject_deferred(cls, data: Any) -> Any:
+ if isinstance(data, dict):
+ if "default_args" in data:
+ raise ValueError("default_args is not supported; use
'templates' with 'extends' instead")
+ for k in DAG_CALLBACK_KEYS:
+ if k in data:
+ raise ValueError(f"Dag-level {k!r} not yet implemented")
+ return data
+
+ @model_validator(mode="before")
+ @classmethod
+ def _resolve_extends(cls, data: Any) -> Any:
+ """Merge in each task's ``extends`` templates before task
validation."""
+ if not isinstance(data, dict):
+ return data
+ templates = data.get("templates")
+ if templates is None:
+ templates = {}
+ tasks = data.get("tasks")
+ # Only merge when the shapes are as expected. A malformed input is left
+ # untouched, so normal field validation reports it as a
ValidationError.
+ if not isinstance(templates, dict) or not isinstance(tasks, list):
+ return data
+ resolved = []
+ for task in tasks:
+ if not isinstance(task, dict) or "extends" not in task:
+ resolved.append(task)
+ continue
+ if not (extends := _validate_extends(task["extends"], where=f"task
{task.get('id')!r}")):
+ resolved.append(task)
+ continue
+ merged: dict = {}
+ for name in extends:
+ merged = _merge_task(merged, _expand_template(name, templates))
+ resolved.append(_merge_task(merged, {k: v for k, v in task.items()
if k != "extends"}))
+ return {**data, "tasks": resolved}
+
+ @model_validator(mode="after")
+ def _check_refs(self):
+ """Check task ID uniqueness and ensure dependencies are internal."""
+ task_id_counts = collections.Counter(t.id_ for t in self.tasks)
+ if duplicates := sorted(i for i, n in task_id_counts.items() if n > 1):
+ raise ValueError(f"duplicate task id(s): {duplicates}")
+ validated_task_ids = set(task_id_counts)
+ for t in self.tasks:
+ refs = set(t.needs)
+ for v in itertools.chain(t.with_.values(), (t.run or {}).values()):
+ refs.update(_iter_xcom_task_ids(v))
+ if t.id_ in refs:
+ raise ValueError(f"task {t.id_!r} depends on itself")
+ if missing := (refs - validated_task_ids):
+ raise ValueError(f"task {t.id_!r} references unknown task(s):
{sorted(missing)}")
+ return self
diff --git a/task-sdk/src/airflow/sdk/importers/yaml_importer/parser.py
b/task-sdk/src/airflow/sdk/importers/yaml_importer/parser.py
new file mode 100644
index 00000000000..13b9fe30c2a
--- /dev/null
+++ b/task-sdk/src/airflow/sdk/importers/yaml_importer/parser.py
@@ -0,0 +1,102 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Parse a YAML/JSON Dag file into validated :class:`~.models.DagDocument`
objects."""
+
+from __future__ import annotations
+
+import collections.abc
+from typing import TYPE_CHECKING, Any
+
+import yaml
+from pydantic import ValidationError
+
+from airflow.sdk.importers.yaml_importer import migrator
+from airflow.sdk.importers.yaml_importer.models import DagDocument
+
+if TYPE_CHECKING:
+ from collections.abc import Hashable, Iterator
+ from typing import IO
+
+ from yaml.nodes import MappingNode
+
+
+class _UniqueKeySafeLoader(yaml.SafeLoader):
+ """
+ A ``SafeLoader`` that rejects duplicate mapping keys.
+
+ PyYAML keeps the last value for a repeated key, so a second ``tasks:``
block
+ or a second ``with:`` or ``needs:`` on a task would silently drop the
first.
+ This can be a footgun for a hand-edited format, so we raise a
+ ``ConstructorError`` instead of silently shadowing.
+ """
+
+ def construct_mapping(self, node: MappingNode, deep: bool = False) ->
dict[Hashable, Any]:
+ seen: set[Hashable] = set()
+ for key_node, _ in node.value:
+ # Let the superclass expand merge keys (<<). It's a valid use case
+ # for the mapping to explicitly override a key supplied by a merge.
+ if key_node.tag == "tag:yaml.org,2002:merge":
+ continue
+ key = self.construct_object(key_node, deep=deep)
+ if not isinstance(key, collections.abc.Hashable):
+ continue
+ if key not in seen:
+ seen.add(key)
+ continue
+ raise yaml.constructor.ConstructorError(
+ "while constructing a mapping",
+ node.start_mark,
+ f"found duplicate key {key!r}",
+ key_node.start_mark,
+ )
+ return super().construct_mapping(node, deep=deep)
+
+
+class YamlDagParseError(ValueError):
+ """A document could not be parsed or did not conform to the format."""
+
+
+def _resolve_and_migrate(raw: dict[str, Any], *, source: str) -> dict[str,
Any]:
+ """Resolve the ``$schema`` version and migrate to the head shape."""
+ if not isinstance(raw, collections.abc.Mapping):
+ raise YamlDagParseError(f"{source}: a Dag document must be a mapping,
got {type(raw).__name__}")
+ if "$schema" not in raw:
+ raise YamlDagParseError(f"{source}: missing required key '$schema'")
+ schema = raw["$schema"]
+ if not isinstance(schema, str) or not schema:
+ raise YamlDagParseError(f"{source}: '$schema' must be a non-empty
string, got {schema!r}")
+ try:
+ return migrator.get_migrator().resolve_and_migrate(raw)
+ except ValueError as exc: # unresolvable $schema version; bundle-config
errors are not ValueError
+ raise YamlDagParseError(f"{source}: {exc}") from exc
+
+
+def parse_documents(stream: str | IO[str], *, source: str = "<string>") ->
Iterator[DagDocument]:
+ """Parse each document of *stream* into a :class:`DagDocument`."""
+ try:
+ for pos, raw in enumerate(yaml.load_all(stream, _UniqueKeySafeLoader)):
+ if raw is None: # empty document between --- separators
+ continue
+ where = source if pos == 0 else f"{source}[doc {pos}]"
+ migrated = _resolve_and_migrate(raw, source=where)
+ try:
+ yield DagDocument.model_validate(migrated)
+ except ValidationError as exc:
+ raise YamlDagParseError(f"{where}: {exc}") from exc
+ except yaml.YAMLError as exc: # a syntax error surfaced while streaming a
document
+ raise YamlDagParseError(f"{source}: invalid YAML: {exc}") from exc
diff --git a/task-sdk/src/airflow/sdk/importers/yaml_importer/schema.json
b/task-sdk/src/airflow/sdk/importers/yaml_importer/schema.json
new file mode 100644
index 00000000000..953c61b6ccc
--- /dev/null
+++ b/task-sdk/src/airflow/sdk/importers/yaml_importer/schema.json
@@ -0,0 +1,365 @@
+{
+ "$defs": {
+ "ConstRef": {
+ "additionalProperties": false,
+ "description": "Force a literal; the value is taken verbatim.",
+ "properties": {
+ "$const": {
+ "title": "$Const"
+ }
+ },
+ "required": [
+ "$const"
+ ],
+ "title": "ConstRef",
+ "type": "object"
+ },
+ "Literal": {
+ "description": "A literal value.",
+ "not": {
+ "not": {
+ "patternProperties": {
+ "^\\$": false
+ }
+ },
+ "type": "object"
+ },
+ "oneOf": [
+ {
+ "additionalProperties": {
+ "oneOf": [
+ {
+ "$ref": "#/$defs/XComRef"
+ },
+ {
+ "$ref": "#/$defs/TemplateRef"
+ },
+ {
+ "$ref": "#/$defs/ConstRef"
+ },
+ {
+ "$ref": "#/$defs/Literal"
+ }
+ ]
+ },
+ "type": "object"
+ },
+ {
+ "items": {
+ "oneOf": [
+ {
+ "$ref": "#/$defs/XComRef"
+ },
+ {
+ "$ref": "#/$defs/TemplateRef"
+ },
+ {
+ "$ref": "#/$defs/ConstRef"
+ },
+ {
+ "$ref": "#/$defs/Literal"
+ }
+ ]
+ },
+ "type": "array"
+ },
+ {
+ "type": "boolean"
+ },
+ {
+ "type": "null"
+ },
+ {
+ "type": "number"
+ },
+ {
+ "type": "string"
+ }
+ ],
+ "title": "Literal"
+ },
+ "Task": {
+ "additionalProperties": true,
+ "description": "A ready-made operator (uses) or task with custom code
(run).",
+ "not": {
+ "allOf": [
+ {
+ "properties": {
+ "uses": {
+ "type": "string"
+ }
+ },
+ "required": [
+ "uses"
+ ]
+ },
+ {
+ "properties": {
+ "run": {
+ "type": "object"
+ }
+ },
+ "required": [
+ "run"
+ ]
+ }
+ ]
+ },
+ "properties": {
+ "id": {
+ "minLength": 1,
+ "title": "Id",
+ "type": "string"
+ },
+ "needs": {
+ "items": {
+ "type": "string"
+ },
+ "title": "Needs",
+ "type": "array"
+ },
+ "extends": {
+ "oneOf": [
+ {
+ "type": "string"
+ },
+ {
+ "items": {
+ "type": "string"
+ },
+ "type": "array"
+ }
+ ]
+ },
+ "with": {
+ "additionalProperties": {
+ "oneOf": [
+ {
+ "$ref": "#/$defs/XComRef"
+ },
+ {
+ "$ref": "#/$defs/TemplateRef"
+ },
+ {
+ "$ref": "#/$defs/ConstRef"
+ },
+ {
+ "$ref": "#/$defs/Literal"
+ }
+ ]
+ },
+ "title": "With",
+ "type": "object"
+ },
+ "uses": {
+ "minLength": 1,
+ "type": [
+ "string",
+ "null"
+ ]
+ },
+ "run": {
+ "additionalProperties": {
+ "oneOf": [
+ {
+ "$ref": "#/$defs/XComRef"
+ },
+ {
+ "$ref": "#/$defs/TemplateRef"
+ },
+ {
+ "$ref": "#/$defs/ConstRef"
+ },
+ {
+ "$ref": "#/$defs/Literal"
+ }
+ ]
+ },
+ "type": [
+ "object",
+ "null"
+ ]
+ }
+ },
+ "required": [
+ "id"
+ ],
+ "title": "Task",
+ "type": "object"
+ },
+ "TaskTemplate": {
+ "additionalProperties": true,
+ "description": "A reusable, partial task fragment merged into any task
that extends it.\n\nThis holds arbitrary keys as-is. All validation is done on
the merged task.",
+ "properties": {},
+ "title": "TaskTemplate",
+ "type": "object"
+ },
+ "TemplateRef": {
+ "additionalProperties": false,
+ "description": "A Jinja template.",
+ "oneOf": [
+ {
+ "required": [
+ "$t"
+ ]
+ },
+ {
+ "required": [
+ "$template"
+ ]
+ }
+ ],
+ "properties": {
+ "$t": {
+ "title": "$T",
+ "type": "string"
+ },
+ "$template": {
+ "title": "$T",
+ "type": "string"
+ }
+ },
+ "title": "TemplateRef",
+ "type": "object"
+ },
+ "TimetableSchedule": {
+ "additionalProperties": false,
+ "description": "A constructed timetable.",
+ "properties": {
+ "uses": {
+ "minLength": 1,
+ "title": "Uses",
+ "type": "string"
+ },
+ "with": {
+ "additionalProperties": true,
+ "title": "With",
+ "type": "object"
+ }
+ },
+ "required": [
+ "uses"
+ ],
+ "title": "TimetableSchedule",
+ "type": "object"
+ },
+ "XComRef": {
+ "additionalProperties": false,
+ "description": "An upstream task's XCom output.",
+ "oneOf": [
+ {
+ "required": [
+ "$x"
+ ]
+ },
+ {
+ "required": [
+ "$xcom"
+ ]
+ }
+ ],
+ "properties": {
+ "$x": {
+ "oneOf": [
+ {
+ "type": "string"
+ },
+ {
+ "$ref": "#/$defs/XComTarget"
+ }
+ ],
+ "title": "$X"
+ },
+ "$xcom": {
+ "oneOf": [
+ {
+ "type": "string"
+ },
+ {
+ "$ref": "#/$defs/XComTarget"
+ }
+ ],
+ "title": "$X"
+ }
+ },
+ "title": "XComRef",
+ "type": "object"
+ },
+ "XComTarget": {
+ "additionalProperties": false,
+ "description": "The object form of an XCom reference.",
+ "properties": {
+ "task": {
+ "title": "Task",
+ "type": "string"
+ },
+ "key": {
+ "default": null,
+ "oneOf": [
+ {
+ "type": "string"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "title": "Key"
+ }
+ },
+ "required": [
+ "task"
+ ],
+ "title": "XComTarget",
+ "type": "object"
+ }
+ },
+ "additionalProperties": true,
+ "description": "A Dag represented by one YAML document.\n\nExtra top-level
keys are Dag-level arguments.",
+ "properties": {
+ "$schema": {
+ "title": "$Schema",
+ "type": "string"
+ },
+ "dag_id": {
+ "minLength": 1,
+ "title": "Dag Id",
+ "type": "string"
+ },
+ "schedule": {
+ "default": null,
+ "oneOf": [
+ {
+ "type": "string"
+ },
+ {
+ "$ref": "#/$defs/TimetableSchedule"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "title": "Schedule"
+ },
+ "templates": {
+ "additionalProperties": {
+ "$ref": "#/$defs/TaskTemplate"
+ },
+ "title": "Templates",
+ "type": "object"
+ },
+ "tasks": {
+ "items": {
+ "$ref": "#/$defs/Task"
+ },
+ "title": "Tasks",
+ "type": "array"
+ }
+ },
+ "required": [
+ "$schema",
+ "dag_id"
+ ],
+ "title": "Airflow YAML Dag (2026-10-30)",
+ "type": "object",
+ "$schema": "https://json-schema.org/draft/2020-12/schema";
+}
diff --git
a/task-sdk/src/airflow/sdk/importers/yaml_importer/versions/__init__.py
b/task-sdk/src/airflow/sdk/importers/yaml_importer/versions/__init__.py
new file mode 100644
index 00000000000..d5259fcf4fc
--- /dev/null
+++ b/task-sdk/src/airflow/sdk/importers/yaml_importer/versions/__init__.py
@@ -0,0 +1,38 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Cadwyn ``VersionBundle`` for the YAML Dag format keyed by ``$schema``.
+
+The ``cadwyn`` import is deferred, so importing this package does not pull in
+FastAPI/Starlette. Versions are newest-first.
+"""
+
+from __future__ import annotations
+
+import functools
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from cadwyn import VersionBundle
+
+
[email protected]
+def get_bundle() -> VersionBundle:
+ """Build the format's ``VersionBundle`` lazily (newest-to-oldest)."""
+ from cadwyn import HeadVersion, Version, VersionBundle
+
+ return VersionBundle(HeadVersion(), Version("2026-10-30"))
diff --git a/task-sdk/tests/task_sdk/importers/yaml_importer/__init__.py
b/task-sdk/tests/task_sdk/importers/yaml_importer/__init__.py
new file mode 100644
index 00000000000..13a83393a91
--- /dev/null
+++ b/task-sdk/tests/task_sdk/importers/yaml_importer/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
diff --git a/task-sdk/tests/task_sdk/importers/yaml_importer/example_dag.yaml
b/task-sdk/tests/task_sdk/importers/yaml_importer/example_dag.yaml
new file mode 100644
index 00000000000..fdd25fa71a6
--- /dev/null
+++ b/task-sdk/tests/task_sdk/importers/yaml_importer/example_dag.yaml
@@ -0,0 +1,69 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Fixture: a realistic two-document YAML Dag for the parser integration test.
+---
+$schema: https://airflow.apache.org/schemas/dag/2026-10-30.json
+dag_id: retail_daily_sales
+schedule: "0 3 * * *"
+start_date: "2025-01-01"
+catchup: false
+tags: [retail, etl]
+params:
+ target_schema: public
+templates:
+ retryable:
+ retries: 2
+ retry_delay: "5m"
+ redshift_load:
+ extends: [retryable]
+ uses:
airflow.providers.amazon.aws.transfers.s3_to_redshift.S3ToRedshiftOperator
+ with:
+ schema: {$t: "{{ params.target_schema }}"}
+ redshift_conn_id: redshift_default
+tasks:
+ - id: wait_for_export
+ extends: [retryable]
+ uses: airflow.providers.standard.sensors.time_delta.TimeDeltaSensor
+ with: {delta: "1h"}
+ - id: load_sales
+ extends: [redshift_load]
+ needs: [wait_for_export]
+ with:
+ table: sales
+ s3_key: {$t: "sales/{{ ds }}.csv"}
+---
+$schema: https://airflow.apache.org/schemas/dag/2026-10-30.json
+dag_id: events_pipeline
+schedule: "@hourly"
+start_date: "2025-01-01"
+catchup: false
+tasks:
+ - id: extract
+ run: {}
+ queue: extract-workers
+ - id: transform
+ run:
+ region: "eu"
+ events: {$x: extract}
+ queue: jvm-workers
+ - id: load
+ uses:
airflow.providers.amazon.aws.transfers.s3_to_redshift.S3ToRedshiftOperator
+ needs: [transform]
+ with:
+ table: events
+ s3_key: {$t: "events/{{ ts_nodash }}.json"}
diff --git a/task-sdk/tests/task_sdk/importers/yaml_importer/test_migrator.py
b/task-sdk/tests/task_sdk/importers/yaml_importer/test_migrator.py
new file mode 100644
index 00000000000..130c33183bb
--- /dev/null
+++ b/task-sdk/tests/task_sdk/importers/yaml_importer/test_migrator.py
@@ -0,0 +1,188 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import pytest
+from cadwyn import (
+ HeadVersion,
+ Version,
+ VersionBundle,
+ VersionChange,
+ convert_request_to_next_version_for,
+ schema,
+)
+
+from airflow.sdk.importers.yaml_importer import migrator
+from airflow.sdk.importers.yaml_importer.models import DagDocument, Task
+
+HEAD_SCHEMA = migrator.schema_from_version("2026-10-30")
+
+
+class _RenameOwnerAttr(VersionChange):
+ """
+ Test-only fake version change.
+
+ This changes an attribute from ``owner_old`` to ``owner``.
+ """
+
+ description = __doc__
+ instructions_to_migrate_to_previous_version = ()
+
+ @convert_request_to_next_version_for(DagDocument) # type: ignore[arg-type]
+ def _move(request):
+ if "owner_old" in request.body:
+ request.body["owner"] = request.body.pop("owner_old")
+
+
+_SYNTH_BUNDLE = VersionBundle(HeadVersion(), Version("2027-06-01",
_RenameOwnerAttr), Version("2026-10-30"))
+
+
+def _migrate(m, date, **extra):
+ body = {"$schema": migrator.schema_from_version(date), "dag_id": "d",
"tasks": [], **extra}
+ return m.resolve_and_migrate(body)
+
+
[email protected](
+ ("schema", "expected"),
+ [
+ pytest.param(migrator.schema_from_version("2026-10-30"), "2026-10-30",
id="canonical-url"),
+ pytest.param("2026-10-30", None, id="bare-date"),
+ pytest.param("garbage-2026-10-30.json", None, id="date-in-junk"),
+ pytest.param("https://mirror.example/schemas/dag/2026-10-30.json";,
None, id="wrong-host"),
+ pytest.param("", None, id="empty"),
+ ],
+)
+def test_version_from_schema_requires_canonical_url(schema, expected):
+ assert migrator.version_from_schema(schema) == expected
+
+
+def test_exact_version_pin_migrates_from_that_version():
+ # Observed through the converter: it runs only when the pinned (exact)
version is older than head.
+ m = migrator.DagDocumentMigrator(_SYNTH_BUNDLE)
+ assert _migrate(m, "2026-10-30", owner_old="t")["owner"] == "t" # exact
older version -> migrates
+ # head-pinned: resolved source == head, so no converter runs
+ at_head = _migrate(m, "2027-06-01", owner_old="t")
+ assert at_head.get("owner_old") == "t"
+ assert "owner" not in at_head
+
+
[email protected](
+ "schema_value",
+ [
+ pytest.param(migrator.schema_from_version("2099-01-01"),
id="newer-than-head"),
+ pytest.param(migrator.schema_from_version("2020-01-01"),
id="older-than-oldest"),
+ pytest.param(migrator.schema_from_version("2027-03-01"),
id="between-versions"),
+ pytest.param("not-a-version", id="junk"),
+ pytest.param("http://airflow.apache.org/schemas/dag/2026-10-30.json";,
id="wrong-scheme"),
+ pytest.param("2026-10-30", id="bare-date"),
+ ],
+)
+def test_unknown_is_rejected(schema_value):
+ m = migrator.DagDocumentMigrator(_SYNTH_BUNDLE)
+ with pytest.raises(ValueError, match=r"not valid|is not a canonical schema
URL"):
+ m.resolve_and_migrate({"$schema": schema_value, "dag_id": "d",
"tasks": []})
+
+
+def test_migrator_rejects_converter_for_non_dagdocument_model():
+ class _ConvertTask(VersionChange):
+ "test-only: a converter keyed on a nested model the migrator does not
drive."
+
+ description = __doc__
+ instructions_to_migrate_to_previous_version = ()
+
+ @convert_request_to_next_version_for(Task) # type: ignore[arg-type]
+ def _noop(request):
+ pass
+
+ bundle = VersionBundle(HeadVersion(), Version("2027-06-01", _ConvertTask),
Version("2026-10-30"))
+ with pytest.raises(RuntimeError, match=r"_ConvertTask:.*only
whole-document DagDocument converters"):
+ migrator.DagDocumentMigrator(bundle)
+
+
+def test_migrator_rejects_non_request_instructions():
+ class _RenameViaSchema(VersionChange):
+ "test-only: a schema instruction the migrator does not apply."
+
+ description = __doc__
+ instructions_to_migrate_to_previous_version = (
+
schema(DagDocument).field("schedule").had(name="schedule_interval"),
+ )
+
+ bundle = VersionBundle(HeadVersion(), Version("2027-06-01",
_RenameViaSchema), Version("2026-10-30"))
+ with pytest.raises(RuntimeError, match="_RenameViaSchema: unsupported
cadwyn instructions"):
+ migrator.DagDocumentMigrator(bundle)
+
+
+def test_known_version_does_not_warn(recwarn):
+ m = migrator.DagDocumentMigrator(_SYNTH_BUNDLE)
+ _migrate(m, "2026-10-30")
+ _migrate(m, "2027-06-01")
+ assert not recwarn.list, "exact known versions must not warn"
+
+
+def test_migrate_is_noop_at_head_for_the_real_bundle():
+ body = {"$schema": HEAD_SCHEMA, "dag_id": "d", "tasks": []}
+ assert migrator.get_migrator().resolve_and_migrate(body) is body # one
version -> no copy
+
+
+class _AppendV2(VersionChange):
+ "test-only: appends a marker so a run is observable and non-idempotent."
+
+ description = __doc__
+ instructions_to_migrate_to_previous_version = ()
+
+ @convert_request_to_next_version_for(DagDocument) # type: ignore[arg-type]
+ def _move(request):
+ request.body["trail"] = request.body.get("trail", "") + "+v2"
+
+
+class _AppendV3(VersionChange):
+ "test-only: appends a marker so a run is observable and non-idempotent."
+
+ description = __doc__
+ instructions_to_migrate_to_previous_version = ()
+
+ @convert_request_to_next_version_for(DagDocument) # type: ignore[arg-type]
+ def _move(request):
+ request.body["trail"] = request.body.get("trail", "") + "+v3"
+
+
+_THREE_VERSIONS = VersionBundle(
+ HeadVersion(),
+ Version("2027-06-01", _AppendV3),
+ Version("2027-01-01", _AppendV2),
+ Version("2026-10-30"),
+)
+
+
+def test_migration_from_middle_version_skips_its_own_converter():
+ # Pinning the middle version applies only the newer converter; the pinned
+ # version's own converter is skipped.
+ m = migrator.DagDocumentMigrator(_THREE_VERSIONS)
+ body = {"$schema": migrator.schema_from_version("2027-01-01"), "dag_id":
"d", "tasks": []}
+ out = m.resolve_and_migrate(body)
+ assert out["trail"] == "+v3"
+ assert "trail" not in body # the input body is not mutated
+
+
+def test_migration_from_oldest_runs_every_newer_converter():
+ m = migrator.DagDocumentMigrator(_THREE_VERSIONS)
+ out = m.resolve_and_migrate(
+ {"$schema": migrator.schema_from_version("2026-10-30"), "dag_id": "d",
"tasks": []}
+ )
+ assert out["trail"] == "+v2+v3" # oldest -> both converters,
oldest-to-newest
diff --git a/task-sdk/tests/task_sdk/importers/yaml_importer/test_parser.py
b/task-sdk/tests/task_sdk/importers/yaml_importer/test_parser.py
new file mode 100644
index 00000000000..22a69df2c75
--- /dev/null
+++ b/task-sdk/tests/task_sdk/importers/yaml_importer/test_parser.py
@@ -0,0 +1,576 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import datetime
+import io
+import json
+import pathlib
+import textwrap
+
+import jsonschema
+import pytest
+import yaml
+
+from airflow.sdk.importers import yaml_importer
+from airflow.sdk.importers.yaml_importer import YamlDagParseError,
parse_documents
+from airflow.sdk.importers.yaml_importer.models import (
+ ConstRef,
+ DagDocument,
+ Literal,
+ Task,
+ TemplateRef,
+ TimetableSchedule,
+ XComRef,
+ XComTarget,
+)
+
+SCHEMA_PATH = pathlib.Path(yaml_importer.__file__).with_name("schema.json")
+
+HEAD = "2026-10-30"
+HEAD_SCHEMA = f"https://airflow.apache.org/schemas/dag/{HEAD}.json";
+
+
+def _one(body: str):
+ return next(iter(parse_documents(f"$schema: {HEAD_SCHEMA}\ndag_id:
d\n{textwrap.dedent(body)}")))
+
+
+def _task(task_yaml: str):
+ return _one("tasks:\n" + textwrap.indent(textwrap.dedent(task_yaml), "
")).tasks[0]
+
+
+def test_literal_by_default():
+ t = _task("- {id: t, run: {region: 'eu', n: 5, flag: true}}")
+ assert isinstance(t.run["region"], Literal)
+ assert t.run["region"].root == "eu"
+ assert t.run["n"].root == 5
+ assert t.run["flag"].root is True
+
+
+def test_date_and_datetime_literals():
+ # Unquoted YAML dates/datetimes are valid literal values; a quoted one
stays a string.
+ t = _task("- {id: t, run: {d: 2026-01-01, dt: 2026-01-01T12:30:00, s:
'2026-01-01'}}")
+ assert t.run["d"].root == datetime.date(2026, 1, 1)
+ assert t.run["dt"].root == datetime.datetime(2026, 1, 1, 12, 30)
+ assert t.run["s"].root == "2026-01-01"
+
+
+def test_xcom_short_long_and_keyed():
+ doc = _one(
+ "tasks:\n - {id: up, run: {}}\n - {id: t, run: {a: {$x: up}, b:
{$xcom: up}, c: {$x: {task: up, key: rows}}}}"
+ )
+ t = {x.id_: x for x in doc.tasks}["t"]
+ assert isinstance(t.run["a"], XComRef)
+ assert t.run["a"].target == "up"
+ assert isinstance(t.run["b"], XComRef)
+ assert t.run["b"].target == "up"
+ assert isinstance(t.run["c"].target, XComTarget)
+ assert t.run["c"].target.key == "rows"
+
+
+def test_template_marker():
+ t = _task("- {id: t, uses: X, with: {k: {$t: 'v-{{ ds }}'}, k2:
{$template: 'x'}}}")
+ assert isinstance(t.with_["k"], TemplateRef)
+ assert "{{ ds }}" in t.with_["k"].source
+ assert isinstance(t.with_["k2"], TemplateRef)
+
+
+def test_const_is_verbatim_not_recursed():
+ t = _task("- {id: t, run: {a: {$const: {$x: not-a-ref}}}}")
+ assert isinstance(t.run["a"], ConstRef)
+ assert t.run["a"].value == {"$x": "not-a-ref"} # inner marker NOT
interpreted
+
+
[email protected](
+ "value",
+ [
+ pytest.param("{$x: up, extra: 1}", id="marker-key-plus-extra"),
+ pytest.param("{$xcomm: up}", id="unknown-marker-typo"),
+ pytest.param("{$tempalte: x}", id="unknown-template-typo"), #
codespell:ignore tempalte
+ pytest.param("{cfg: {$weird: 1}}", id="dollar-key-nested-in-literal"),
+ ],
+)
+def test_reserved_dollar_keys_rejected(value):
+ # A '$'-prefixed key that is not a recognised sole-key marker is rejected,
not silently literal.
+ with pytest.raises(YamlDagParseError, match="reserved"):
+ _task(f"- {{id: t, run: {{a: {value}}}}}")
+
+
+def test_nested_marker_inside_literal_resolved():
+ t = _task("- {id: t, uses: X, with: {cfg: {url: {$t: '{{ ds }}'}, n: 1}}}")
+ inner = t.with_["cfg"].root
+ assert isinstance(inner["url"], TemplateRef)
+ assert inner["n"].root == 1
+
+
+def test_operator_discrimination():
+ op = _task("- {id: t, uses: a.b.C, with: {x: 1}}")
+ assert isinstance(op, Task)
+ assert op.uses == "a.b.C"
+ assert op.run is None
+
+
+def test_code_discrimination():
+ code = _task("- {id: t, run: {}}")
+ assert isinstance(code, Task)
+ assert code.run == {}
+ assert code.uses is None
+
+
+def test_no_body_is_allowed():
+ t = _task("- {id: t}")
+ assert t.uses is None
+ assert t.run is None
+
+
+def test_both_bodies_is_error():
+ with pytest.raises(YamlDagParseError, match="may not set both 'uses' and
'run'"):
+ _task("- {id: t, uses: X, run: {}}")
+
+
[email protected](
+ "task",
+ [
+ pytest.param("{id: t, uses: null}", id="uses-null"),
+ pytest.param("{id: t, run: null}", id="run-null"),
+ pytest.param("{id: t, uses: null, run: null}", id="both-null"),
+ pytest.param("{id: t, uses: X, run: null}", id="uses-with-run-null"),
+ ],
+)
+def test_null_body_is_allowed(task):
+ # An explicit null body means "no body" (like omission), so a generator
may emit it.
+ t = _task(f"- {task}")
+ assert t.run is None
+
+
+def test_empty_uses_is_rejected():
+ # An empty string is not a valid import path; null is the way to say "no
body".
+ with pytest.raises(YamlDagParseError, match="at least 1 character"):
+ _task('- {id: t, uses: ""}')
+
+
[email protected](
+ "doc",
+ [
+ pytest.param("dag_id: ''\ntasks: []", id="dag-id"),
+ pytest.param("dag_id: d\ntasks:\n - {id: '', run: {}}", id="task-id"),
+ pytest.param("dag_id: d\nschedule: {uses: '', with: {}}\ntasks: []",
id="timetable-uses"),
+ ],
+)
+def test_empty_identifiers_rejected(doc):
+ # Identifiers must be non-empty here; Airflow's validate_key enforces the
full
+ # charset/length rules later, when the importer builds the Dag.
+ with pytest.raises(YamlDagParseError, match="at least 1 character"):
+ list(parse_documents(f"$schema: {HEAD_SCHEMA}\n{doc}"))
+
+
+def test_python_field_names_not_accepted_as_aliases():
+ with pytest.raises(YamlDagParseError):
+ _one("tasks:\n - {id_: t, run: {}}") # 'id_' is not 'id' -> id
missing
+ t = _task("- {id: t, uses: X, with_: {k: 1}}")
+ assert t.with_ == {} # 'with_' did not populate the 'with' field
+ assert t.model_extra.get("with_") == {"k": 1} # it is just a pass-through
extra
+
+
+def test_default_args_rejected():
+ with pytest.raises(YamlDagParseError, match="default_args"):
+ _one("default_args: {retries: 1}\ntasks: []")
+
+
+def test_callbacks_rejected():
+ with pytest.raises(YamlDagParseError, match="callback"):
+ _one("on_failure_callback: cb\ntasks: []")
+ with pytest.raises(YamlDagParseError, match="callback"):
+ _task("- {id: t, run: {}, on_success_callback: cb}")
+
+
+def test_inlets_outlets_rejected():
+ with pytest.raises(YamlDagParseError, match="not yet implemented"):
+ _task("- {id: t, run: {}, outlets: [a]}")
+
+
+def test_extends_merges_shallow_task_wins():
+ doc = _one(
+ """
+ templates:
+ retryable: {retries: 2, retry_delay: '5m'}
+ load:
+ extends: [retryable]
+ uses: a.b.S3ToRedshiftOperator
+ with: {schema: public, conn_id: warehouse}
+ tasks:
+ - id: t
+ extends: [load]
+ with: {table: sales, conn_id: override}
+ """
+ )
+ t = doc.tasks[0]
+ assert isinstance(t, Task)
+ assert t.uses.endswith("S3ToRedshiftOperator")
+ assert t.model_extra["retries"] == 2 # from retryable
+ assert isinstance(t.with_["schema"], Literal) # inherited
+ assert t.with_["table"].root == "sales" # own
+ assert t.with_["conn_id"].root == "override" # one-level with-merge, task
wins
+
+
+def test_unknown_template_errors():
+ with pytest.raises(YamlDagParseError, match="unknown template"):
+ _one("tasks:\n - {id: t, extends: [ghost], run: {}}")
+
+
+def test_template_cycle_errors():
+ with pytest.raises(YamlDagParseError, match="cycle"):
+ _one("templates: {a: {extends: [b]}, b: {extends: [a]}}\ntasks: [{id:
t, extends: [a], run: {}}]")
+
+
[email protected](
+ "body",
+ [
+ pytest.param("tasks:\n - {id: t, run: {}, extends: [[base]]}",
id="task-extends-nested-list"),
+ pytest.param("tasks:\n - {id: t, run: {}, extends: [{base: 1}]}",
id="task-extends-dict-element"),
+ pytest.param("tasks:\n - {id: t, run: {}, extends: 1}",
id="task-extends-non-string"),
+ ],
+)
+def test_task_extends_must_be_a_name_or_list_of_names(body):
+ # A non-string entry (or a non-string/non-list value) is a parse error,
not a TypeError.
+ with pytest.raises(YamlDagParseError, match="extends"):
+ _one(body)
+
+
+def test_template_extends_must_be_a_name_or_list_of_names():
+ with pytest.raises(YamlDagParseError, match="extends"):
+ _one(
+ "templates: {base: {retries: 1}, child: {extends:
[[base]]}}\ntasks: [{id: t, extends: [child], run: {}}]"
+ )
+
+
+def test_extends_accepts_a_bare_string():
+ # `extends: base` is shorthand for `extends: [base]`, at both task and
template level.
+ doc = _one(
+ "templates:\n gp: {retries: 3}\n base: {extends: gp}\ntasks:\n -
{id: t, extends: base, uses: X}"
+ )
+ assert doc.tasks[0].model_extra["retries"] == 3 # merged gp -> base ->
task
+
+
+def test_unknown_needs_rejected():
+ with pytest.raises(YamlDagParseError, match="unknown task"):
+ _one("tasks:\n - {id: t, run: {}, needs: [ghost]}")
+
+
[email protected](
+ "task",
+ [
+ pytest.param("{id: t, run: {e: {$x: ghost}}}", id="run-top-level"),
+ pytest.param("{id: t, run: {e: [{$x: ghost}]}}", id="run-in-list"),
+ pytest.param("{id: t, run: {e: {nested: {$x: ghost}}}}",
id="run-nested-dict"),
+ pytest.param("{id: t, uses: X, with: {a: {$x: ghost}}}",
id="with-on-operator"),
+ ],
+)
+def test_unknown_xcom_target_rejected(task):
+ # XCom refs must be found at any depth inside with/run (the list and
nested-dict branches too).
+ with pytest.raises(YamlDagParseError, match="unknown task"):
+ _one(f"tasks:\n - {task}")
+
+
+def test_duplicate_task_ids_rejected():
+ with pytest.raises(YamlDagParseError, match="duplicate task id"):
+ _one("tasks:\n - {id: a, run: {}}\n - {id: a, uses: x.Y}")
+
+
[email protected](
+ "body",
+ [
+ pytest.param("tasks:\n - {id: a, run: {}, needs: [a]}",
id="self-needs"),
+ pytest.param("tasks:\n - {id: a, run: {x: {$x: a}}}", id="self-xcom"),
+ ],
+)
+def test_self_dependency_rejected(body):
+ with pytest.raises(YamlDagParseError, match="depends on itself"):
+ _one(body)
+
+
+def test_schedule_forms():
+ assert _one("schedule: '0 3 * * *'\ntasks: []").schedule == "0 3 * * *"
+ assert _one("schedule: null\ntasks: []").schedule is None
+ tt = _one("schedule: {uses: a.b.MyTimetable, with: {n: 1}}\ntasks:
[]").schedule
+ assert isinstance(tt, TimetableSchedule)
+ assert tt.uses.endswith("MyTimetable")
+
+
+def test_extra_top_level_keys_pass_through():
+ # Unknown top-level keys are Dag-level arguments: accepted and surfaced
via pydantic extras.
+ doc = _one("catchup: false\nmax_active_runs: 3\ntags: [etl]\ntasks: []")
+ assert doc.model_extra == {"catchup": False, "max_active_runs": 3, "tags":
["etl"]}
+
+
+def test_schema_required():
+ with pytest.raises(YamlDagParseError, match=r"\$schema"):
+ list(parse_documents("dag_id: d\ntasks: []"))
+
+
[email protected](
+ "schema_value",
+ [
+ pytest.param("null", id="null"),
+ pytest.param('""', id="empty-string"),
+ pytest.param("false", id="false"),
+ pytest.param("123", id="int"),
+ pytest.param("[a, b]", id="list"),
+ ],
+)
+def test_schema_must_be_a_non_empty_string(schema_value):
+ # A present-but-falsy or wrong-typed $schema is reported as such, not as
"missing".
+ with pytest.raises(YamlDagParseError, match="must be a non-empty string"):
+ list(parse_documents(f"$schema: {schema_value}\ndag_id: d\ntasks: []"))
+
+
[email protected](
+ "body",
+ [
+ pytest.param("tasks: []\ntasks: []", id="dup-tasks-top-level"),
+ pytest.param("tasks:\n - {id: a, uses: X, with: {x: 1}, with: {y:
2}}", id="dup-with-on-task"),
+ pytest.param("tasks:\n - {id: a, run: {}, needs: [b], needs: [c]}",
id="dup-needs-on-task"),
+ ],
+)
+def test_duplicate_keys_rejected(body):
+ # PyYAML keeps the last value for a repeated key; a hand-edited format
should reject it instead
+ # of silently dropping the first. The loader raises a ConstructorError (a
YAMLError).
+ with pytest.raises(YamlDagParseError, match="duplicate key"):
+ _one(body)
+
+
+def test_yaml_merge_keys_supported():
+ # ``<<`` merge keys resolve via flatten_mapping; the duplicate-key loader
must skip the merge
+ # tag rather than choke on it. Explicit keys still win over merged ones,
and genuine duplicate
+ # explicit keys are still rejected (covered by
test_duplicate_keys_rejected).
+ t = _task(
+ """
+ - id: a
+ run:
+ <<: &defaults {retries: 2, region: eu}
+ region: us
+ """
+ )
+ assert t.run["retries"].root == 2 # pulled in from the merge source
+ assert t.run["region"].root == "us" # explicit key overrides the merged
value
+
+
[email protected](
+ "date",
+ [
+ pytest.param("2099-01-01", id="newer-than-head"),
+ pytest.param("2000-01-01", id="older-than-oldest"),
+ ],
+)
+def test_unknown_schema_version_is_rejected(date):
+ url = f"https://airflow.apache.org/schemas/dag/{date}.json";
+ with pytest.raises(YamlDagParseError, match="not valid"):
+ list(parse_documents(f"$schema: {url}\ndag_id: d\ntasks: []"))
+
+
[email protected](
+ "schema_value",
+ [
+ pytest.param("nonsense", id="junk"),
+ pytest.param("http://airflow.apache.org/schemas/dag/2026-10-30.json";,
id="wrong-scheme"),
+ pytest.param("https://airflow.apache.org/schemas/dag/2026-10-30";,
id="missing-json"),
+ pytest.param("'2026-10-30'", id="quoted-date"),
+ ],
+)
+def test_non_canonical_schema_url_is_rejected(schema_value):
+ # A non-canonical $schema names the value written and the expected form,
not "version None".
+ with pytest.raises(YamlDagParseError, match="is not a canonical schema
URL"):
+ list(parse_documents(f"$schema: {schema_value}\ndag_id: d\ntasks: []"))
+
+
+def test_accepts_a_file_like_stream():
+ stream = io.StringIO(f"$schema: {HEAD_SCHEMA}\ndag_id: d\ntasks: [{{id: t,
run: {{}}}}]")
+ docs = list(parse_documents(stream))
+ assert [d.dag_id for d in docs] == ["d"]
+
+
+def test_parse_is_lazy_errors_surface_on_iteration():
+ # Building the iterator does not parse; the error only surfaces when
consumed.
+ gen = parse_documents("dag_id: d\ntasks: []") # missing $schema
+ with pytest.raises(YamlDagParseError, match=r"\$schema"):
+ next(iter(gen))
+
+
+def test_multiple_documents():
+ docs = parse_documents(
+ f"$schema: {HEAD_SCHEMA}\ndag_id: a\ntasks: []\n---\n$schema:
{HEAD_SCHEMA}\ndag_id: b\ntasks: []\n"
+ )
+ assert [d.dag_id for d in docs] == ["a", "b"]
+
+
+def test_non_mapping_document_rejected():
+ with pytest.raises(YamlDagParseError, match="must be a mapping"):
+ list(parse_documents("- a\n- b"))
+
+
+def test_invalid_yaml_rejected():
+ with pytest.raises(YamlDagParseError, match="invalid YAML"):
+ list(parse_documents("key: [unclosed"))
+
+
+def test_error_label_points_at_offending_document():
+ # A failure in a later document of a multi-doc stream is labelled with its
position.
+ good = f"$schema: {HEAD_SCHEMA}\ndag_id: a\ntasks: []"
+ bad = "dag_id: b\ntasks: []" # missing $schema -> fails, and it is doc
index 1
+ with pytest.raises(YamlDagParseError, match=r"\[doc 1\]"):
+ list(parse_documents(f"{good}\n---\n{bad}"))
+
+
+def test_model_json_schema_describes_markers():
+ # The models are the schema source of truth (the prek dump script
decorates + snapshots
+ # them). Assert the generated schema shape here; the snapshot itself is
enforced by the
+ # generate-yaml-importer-schema-snapshot prek hook.
+ sch = DagDocument.model_json_schema(by_alias=True)
+ assert set(sch["properties"]) >= {"$schema", "dag_id", "schedule",
"tasks", "templates"}
+ assert {"XComRef", "TemplateRef", "ConstRef", "Task"} <= set(sch["$defs"])
+
+
+def test_parses_example_fixture():
+ fixture = pathlib.Path(__file__).parent / "example_dag.yaml"
+ with fixture.open(encoding="utf-8") as fh:
+ docs = list(parse_documents(fh, source=str(fixture)))
+ assert [d.dag_id for d in docs] == ["retail_daily_sales",
"events_pipeline"]
+
+ d1 = {t.id_: t for t in docs[0].tasks}
+ load = d1["load_sales"]
+ assert isinstance(load, Task)
+ assert load.uses.endswith("S3ToRedshiftOperator")
+ assert load.model_extra["retries"] == 2 # via redshift_load -> retryable
+ assert isinstance(load.with_["schema"], TemplateRef) # inherited $t
+ assert isinstance(load.with_["s3_key"], TemplateRef) # own $t
+ assert load.needs == ["wait_for_export"]
+ assert docs[0].model_extra["catchup"] is False
+
+ d2 = {t.id_: t for t in docs[1].tasks}
+ assert isinstance(d2["extract"], Task)
+ assert d2["extract"].run == {}
+ assert d2["extract"].queue == "extract-workers"
+ assert isinstance(d2["transform"].run["events"], XComRef) # {$x: extract}
edge
+ assert d2["load"].needs == ["transform"]
+
+
+def test_example_fixture_conforms_to_published_schema():
+ # Separately validate the fixture against the JSON Schema. Not all Pydantic
+ # features survive the export.
+ validator =
jsonschema.Draft202012Validator(json.loads(SCHEMA_PATH.read_text()))
+ fixture = pathlib.Path(__file__).parent / "example_dag.yaml"
+ documents = [d for d in yaml.safe_load_all(fixture.read_text()) if d]
+ assert documents
+ for document in documents:
+ assert list(validator.iter_errors(document)) == []
+
+
[email protected](
+ "bad_task",
+ [
+ pytest.param({"id": "t", "uses": "a.b.C", "run": {}},
id="both-bodies"),
+ pytest.param({"id": "t", "uses": ""}, id="empty-uses"),
+ ],
+)
+def test_published_schema_rejects_invalid_tasks(bad_task):
+ validator =
jsonschema.Draft202012Validator(json.loads(SCHEMA_PATH.read_text()))
+ document = {"$schema": HEAD_SCHEMA, "dag_id": "d", "tasks": [bad_task]}
+ assert list(validator.iter_errors(document))
+
+
[email protected](
+ "task",
+ [
+ pytest.param({"id": "t"}, id="no-body"),
+ pytest.param({"id": "t", "uses": None}, id="uses-null"),
+ pytest.param({"id": "t", "run": None}, id="run-null"),
+ pytest.param({"id": "t", "uses": "a.b.C", "run": None},
id="uses-with-run-null"),
+ ],
+)
+def test_published_schema_accepts_null_and_empty_bodies(task):
+ validator =
jsonschema.Draft202012Validator(json.loads(SCHEMA_PATH.read_text()))
+ document = {"$schema": HEAD_SCHEMA, "dag_id": "d", "tasks": [task]}
+ assert list(validator.iter_errors(document)) == []
+
+
[email protected](
+ "body",
+ [
+ pytest.param(
+ "tasks:\n - {id: up, run: {}}\n - {id: t, run: {a: {$x: up}}}",
+ id="xcom-short",
+ ),
+ pytest.param(
+ "tasks:\n - {id: up, run: {}}\n - {id: t, run: {a: {$xcom:
up}}}",
+ id="xcom-long",
+ ),
+ pytest.param(
+ "tasks:\n - {id: up, run: {}}\n - {id: t, run: {a: {$x: {task:
up, key: rows}}}}",
+ id="xcom-keyed",
+ ),
+ pytest.param(
+ "tasks:\n - {id: t, run: {a: {$const: {$x: literal}}}}",
+ id="const-around-dollar",
+ ),
+ pytest.param(
+ "tasks:\n - {id: t, uses: X, with: {k: {$t: 'v-{{ ds }}'}}}",
+ id="template",
+ ),
+ pytest.param(
+ "templates: {base: {run: {r: 1}}}\ntasks:\n - {id: t, extends:
base}",
+ id="extends-bare",
+ ),
+ pytest.param(
+ "tasks:\n - {id: t, run: {a: {}}}",
+ id="empty-dict-value",
+ ),
+ pytest.param(
+ "tasks:\n - {id: t, run: {s: x, n: 5, f: 1.5, b: true, z: null}}",
+ id="scalars",
+ ),
+ ],
+)
+def test_valid_shapes_round_trip(body):
+ # Each valid value-grammar shape must be accepted by BOTH the parser and
the published
+ # JSON Schema -- the two descriptions of the format must not drift apart.
+ text = f"$schema: {HEAD_SCHEMA}\ndag_id: d\n{textwrap.dedent(body)}"
+ assert list(parse_documents(text)) # the parser accepts it
+ validator =
jsonschema.Draft202012Validator(json.loads(SCHEMA_PATH.read_text()))
+ for document in (d for d in yaml.safe_load_all(text) if d):
+ assert list(validator.iter_errors(document)) == [] # the published
schema agrees
+
+
[email protected](
+ "value",
+ [
+ pytest.param("{$x: up, extra: 1}", id="marker-plus-extra"),
+ pytest.param("{$xcomm: up}", id="unknown-marker-typo"),
+ pytest.param("{$tempalte: x}", id="unknown-template-typo"), #
codespell:ignore tempalte
+ pytest.param("{cfg: {$weird: 1}}", id="dollar-key-nested"),
+ pytest.param("{$x: up, $xcom: up}", id="two-marker-keys"),
+ pytest.param("{$x: {task: up, nope: 1}}", id="xcom-target-extra-key"),
+ ],
+)
+def test_invalid_shapes_rejected_by_both(value):
+ # Hand-written schema guards (Literal '$'-key 'not', XComTarget and markers
+ # additionalProperties: false) must reject the same shapes as the parser.
+ text = f"$schema: {HEAD_SCHEMA}\ndag_id: d\ntasks:\n - {{id: t, run: {{a:
{value}}}}}"
+ with pytest.raises(YamlDagParseError):
+ list(parse_documents(text)) # the parser rejects it
+ validator =
jsonschema.Draft202012Validator(json.loads(SCHEMA_PATH.read_text()))
+ assert list(validator.iter_errors(yaml.safe_load(text))) # the published
schema agrees
