Add Airflow UI plugin: Blueprint tab showing YAML and blueprint code

README.md

@@ -250,6 +250,19 @@ Every task instance gets two extra fields visible in Airflow's "Rendered Templat
 
 This makes it easy to understand what generated each task instance without leaving the Airflow UI.
 
+## Airflow UI Plugin (Airflow 3)
+
+Installing `airflow-blueprint` registers an Airflow plugin that puts the YAML front and center in the UI. Airflow's built-in Code tab shows the loader file that built the DAG — the plugin adds a **Blueprint** tab (Airflow 3.1+) that shows the YAML and blueprint Python instead:
+
+- On a DAG or DAG run: the source YAML the DAG was built from, with the code of each blueprint it uses underneath.
+- On a task or task instance: that step's config and its blueprint source. Where Airflow stored the run's rendered fields, the config is the exact one the task ran with — param overrides included — falling back to the serialized DAG and then the file on disk, labeled with which you're seeing.
+
+The pages are served by a FastAPI app mounted at `/blueprint` under the API server (Airflow 3.0+), so you can also open them directly.
+
+To map a DAG back to its YAML, `build_all_airflow_dags()` tags each DAG with `blueprint:<path relative to the dags folder>` and records the path in each task's `blueprint_step_config`. Both survive DAG serialization, so the plugin resolves the file with one lookup instead of re-scanning the dags folder (it still falls back to a scan for DAGs built by older versions). Pass `source_tags=False` to skip the tag.
+
+On Airflow 2 the plugin loads but registers nothing.
+
 ## Runtime Parameter Overrides
 
 Blueprints that set `supports_params = True` have their config fields registered as Airflow [DAG params](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/params.html), namespaced as `{step}__{field}`. When you trigger a DAG from the Airflow UI, the trigger form shows those fields pre-filled with YAML defaults — users can override any value before running.

blueprint/builder.py

@@ -30,6 +30,9 @@
 
 DEFAULT_START_DATE = datetime(2024, 1, 1, tzinfo=timezone.utc)
 
+SOURCE_TAG_PREFIX = "blueprint:"
+_TAG_MAX_LEN = 100
+
 _PARAM_SCHEMA_KEYS = frozenset(
     {
         "type",
@@ -186,11 +189,14 @@ def _ensure_start_date(dag_kwargs: dict[str, Any]) -> None:
                 parsed = parsed.replace(tzinfo=timezone.utc)
             dag_kwargs["start_date"] = parsed
 
-    def build(self, config: DAGConfig) -> "DAG":
+    def build(self, config: DAGConfig, source: str | None = None) -> "DAG":
         """Build a DAG from a DAGConfig.
 
         Args:
             config: The parsed and validated DAG configuration
+            source: Path of the YAML file this DAG was built from, recorded
+                in each task's ``blueprint_step_config`` so the source
+                survives DAG serialization.
 
         Returns:
             A fully wired Airflow DAG
@@ -235,7 +241,7 @@ def build(self, config: DAGConfig) -> "DAG":
 
         with dag:
             for step_name, step_config in config.steps.items():
-                rendered[step_name] = self._render_step(step_name, step_config)
+                rendered[step_name] = self._render_step(step_name, step_config, source)
 
             for step_name, step_config in config.steps.items():
                 for dep_name in step_config.depends_on:
@@ -274,7 +280,7 @@ def build_from_yaml(
             raw_config = yaml.safe_load(raw_content)
 
         dag_config = DAGConfig.model_validate(raw_config)
-        dag = self.build(dag_config)
+        dag = self.build(dag_config, source=str(yaml_path))
 
         if self._on_dag_built:
             self._on_dag_built(dag, yaml_path)
@@ -359,6 +365,7 @@ def _render_step(
         self,
         step_name: str,
         step_config: StepConfig,
+        source: str | None = None,
     ) -> TaskOrGroup:
         """Render a single step by instantiating its blueprint."""
         from pydantic import ValidationError
@@ -411,15 +418,14 @@ def _render_step(
 
         result = instance.render(validated_config)
 
-        step_yaml = yaml.dump(
-            {
-                "blueprint": step_config.blueprint,
-                "version": resolved_version,
-                **blueprint_config,
-            },
-            default_flow_style=False,
-            sort_keys=False,
-        )
+        step_context: dict[str, Any] = {
+            "blueprint": step_config.blueprint,
+            "version": resolved_version,
+        }
+        if source is not None:
+            step_context["source"] = source
+        step_context.update(blueprint_config)
+        step_yaml = yaml.dump(step_context, default_flow_style=False, sort_keys=False)
 
         source_code = bp_class.get_source_code()
 
@@ -495,6 +501,33 @@ def _check_duplicate_dag_id(dag_id: str, yaml_path: Path, dag_id_to_file: dict[s
         raise DuplicateDAGIdError(dag_id, [dag_id_to_file[dag_id], yaml_path])
 
 
+def _relative_source(yaml_path: Path, search_path: Path) -> str:
+    """Return yaml_path relative to search_path, or as-is if not beneath it."""
+    try:
+        return str(yaml_path.relative_to(search_path))
+    except ValueError:
+        return str(yaml_path)
+
+
+def _apply_source_tag(dag: "DAG", source: str) -> None:
+    """Tag a DAG with its source YAML path (``blueprint:<path>``).
+
+    The tag survives DAG serialization, so the Airflow UI plugin can map a
+    dag_id back to its YAML file without re-scanning the dags folder. Skipped
+    with a warning if the tag would exceed Airflow's tag length limit.
+    """
+    tag = f"{SOURCE_TAG_PREFIX}{source}"
+    if len(tag) > _TAG_MAX_LEN:
+        logger.warning(
+            "Skipping source tag for DAG '%s': '%s' exceeds %d characters",
+            dag.dag_id,
+            tag,
+            _TAG_MAX_LEN,
+        )
+        return
+    dag.tags = {*(dag.tags or ()), tag}
+
+
 def build_all_airflow_dags(
     search_path: str | Path | None = None,
     register_globals: dict | None = None,
@@ -503,6 +536,7 @@ def build_all_airflow_dags(
     template_context: dict[str, Any] | None = None,
     bp_registry: BlueprintRegistry | None = None,
     on_dag_built: OnDagBuilt | None = None,
+    source_tags: bool = True,
 ) -> list["DAG"]:
     """Discover and build all DAGs from YAML files.
 
@@ -530,6 +564,9 @@ def build_all_airflow_dags(
         on_dag_built: Optional callback invoked after each DAG is built.
             Receives the DAG and the Path to the source YAML file.
             Use this to apply post-processing such as access controls or tags.
+        source_tags: Whether to tag each DAG with its source YAML path
+            (``blueprint:<path relative to search_path>``). The tag lets the
+            Airflow UI plugin resolve a DAG back to its YAML file.
 
     Returns:
         List of built DAGs
@@ -583,7 +620,11 @@ def build_all_airflow_dags(
 
         dag_config = DAGConfig.model_validate(raw_config)
         _check_duplicate_dag_id(dag_config.dag_id, yaml_path, dag_id_to_file)
-        dag = builder.build(dag_config)
+        source = _relative_source(yaml_path, resolved_path)
+        dag = builder.build(dag_config, source=source)
+
+        if source_tags:
+            _apply_source_tag(dag, source)
 
         if on_dag_built:
             on_dag_built(dag, yaml_path)

blueprint/plugin/__init__.py

@@ -0,0 +1,98 @@
+"""Airflow UI plugin: the YAML and blueprint code behind each DAG.
+
+Registered automatically via the ``airflow.plugins`` entry point when
+airflow-blueprint is installed. On Airflow 3 it mounts a FastAPI app at
+``/blueprint`` under the API server; on Airflow 3.1+ it adds a
+"Blueprint" tab at every level of the UI:
+
+- DAG and DAG-run pages show the source YAML the DAG was built from,
+  with the blueprint Python underneath
+- task and task-instance pages show that step's config and blueprint
+  source, resolved to what actually ran where possible
+
+On Airflow 2.x the plugin loads but registers nothing.
+"""
+
+import logging
+from typing import Any
+
+from airflow.plugins_manager import AirflowPlugin
+
+logger = logging.getLogger(__name__)
+
+URL_PREFIX = "/blueprint"
+
+
+def _airflow_version() -> tuple[int, int]:
+    """Return the running Airflow (major, minor), or (0, 0) if unknown."""
+    try:
+        from airflow import __version__ as airflow_version
+        from packaging.version import Version
+
+        v = Version(airflow_version)
+    except Exception:
+        return (0, 0)
+    return (v.major, v.minor)
+
+
+def _build_surfaces() -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    """Build the fastapi_apps and external_views lists for this Airflow version."""
+    version = _airflow_version()
+    if version < (3, 0):
+        return [], []
+
+    try:
+        from blueprint.plugin.app import create_app
+
+        app = create_app()
+    except ImportError as e:
+        logger.warning("Blueprint UI plugin disabled: %s", e)
+        return [], []
+
+    fastapi_apps = [{"app": app, "url_prefix": URL_PREFIX, "name": "Blueprint"}]
+
+    if version < (3, 1):
+        return fastapi_apps, []
+
+    external_views = [
+        {
+            "name": "Blueprint",
+            "href": f"{URL_PREFIX}/dags/{{DAG_ID}}/yaml",
+            "destination": "dag",
+            "url_route": "blueprint_dag",
+        },
+        {
+            "name": "Blueprint",
+            "href": f"{URL_PREFIX}/dags/{{DAG_ID}}/yaml",
+            "destination": "dag_run",
+            "url_route": "blueprint_dag_run",
+        },
+        {
+            "name": "Blueprint",
+            "href": f"{URL_PREFIX}/dags/{{DAG_ID}}/tasks/{{TASK_ID}}",
+            "destination": "task",
+            "url_route": "blueprint_task",
+        },
+        {
+            "name": "Blueprint",
+            "href": (
+                f"{URL_PREFIX}/dags/{{DAG_ID}}/tasks/{{TASK_ID}}"
+                "?run_id={RUN_ID}&map_index={MAP_INDEX}"
+            ),
+            "destination": "task_instance",
+            "url_route": "blueprint_task_instance",
+        },
+    ]
+    return fastapi_apps, external_views
+
+
+_fastapi_apps, _external_views = _build_surfaces()
+
+
+class BlueprintPlugin(AirflowPlugin):
+    """Airflow plugin exposing Blueprint's YAML source and catalog views."""
+
+    name = "blueprint"
+
+    fastapi_apps = _fastapi_apps
+    external_views = _external_views