Implement AIP-85 YAML importer based on the parser

task-sdk/pyproject.toml

@@ -54,14 +54,15 @@ dependencies = [
     "babel>=2.17.0",
     "cadwyn>=6.1.1",
     "fsspec>=2023.10.0",
+    "greenback>=1.2.1",
     "httpx>=0.27.0",
     "jinja2>=3.1.5",
     "methodtools>=0.4.7",
     "msgspec>=0.19.0",
-    "python-dateutil>=2.7.0",
     "psutil>=6.1.0",
+    "python-dateutil>=2.7.0",
+    "pyyaml>=6.0.3",
     "structlog>=26.1.0",
-    "greenback>=1.2.1",
     "tenacity>=8.3.0",
     # Start of shared timezones dependencies
     "pendulum>=3.1.0",

task-sdk/src/airflow/sdk/importers/base.py

@@ -34,11 +34,17 @@
 
 if TYPE_CHECKING:
     from collections.abc import Generator, Iterable, Iterator
-    from typing import Self
+    from typing import Protocol, Self
 
     from airflow.dag_processing.bundles.base import BaseDagBundle  # noqa: SDK002
     from airflow.sdk import DAG
 
+    class _SiblingResource(Protocol):
+        """A readable handle to a resource, e.g. a :class:`pathlib.Path` or :class:`zipfile.Path`."""
+
+        def read_bytes(self) -> bytes: ...
+
+
 log = logging.getLogger(__name__)
 
 DefT = TypeVar("DefT", bound="DagDefinition")
@@ -64,6 +70,22 @@ def read_text(self, encoding: str = "utf-8") -> str:
         """Read and return the content of the resource as a string."""
         return self.read_bytes().decode(encoding)
 
+    def sibling(self, relpath: str) -> _SiblingResource:
+        """
+        Return a readable handle to a resource stored next to this definition.
+
+        The handle supports ``read_bytes()``. Resolution is relative to the
+        directory of the definition's own file and raises if *relpath* escapes
+        it. This is close to a Python Dag's default Jinja search path, but
+        stricter: a path escaping via a symlink is rejected (Jinja's
+        ``FileSystemLoader`` would allow it), and a Dag-level
+        ``template_searchpath`` is not consulted.
+
+        The base cannot locate siblings; a definition type that can hold them
+        overrides this.
+        """
+        raise NotImplementedError(f"{type(self).__name__} cannot resolve sibling resources")
+
     @abstractmethod
     def as_file(self) -> contextlib.AbstractContextManager[Path]:
         """
@@ -133,6 +155,13 @@ def suffix(self) -> str:
     def read_bytes(self) -> bytes:
         return self.path.read_bytes()
 
+    def sibling(self, relpath: str) -> Path:
+        base = self.path.parent.resolve()
+        target = base.joinpath(relpath).resolve()
+        if not target.is_relative_to(base):
+            raise ValueError(f"{relpath!r} resolves outside the Dag file's directory {str(base)!r}")
+        return target
+
     @contextlib.contextmanager
     def as_file(self) -> Generator[Path, None, None]:
         yield self.path

task-sdk/src/airflow/sdk/importers/yaml_importer/__init__.py

@@ -19,14 +19,15 @@
 
 from __future__ import annotations
 
+from airflow.sdk.importers.yaml_importer.builder import YamlDagBuildError
+from airflow.sdk.importers.yaml_importer.importer import YamlDagImporter
 from airflow.sdk.importers.yaml_importer.models import DagDocument
-from airflow.sdk.importers.yaml_importer.parser import (
-    YamlDagParseError,
-    parse_documents,
-)
+from airflow.sdk.importers.yaml_importer.parser import YamlDagParseError, parse_documents
 
 __all__ = [
     "DagDocument",
+    "YamlDagBuildError",
+    "YamlDagImporter",
     "YamlDagParseError",
     "parse_documents",
 ]