Migrate to FastAPI lazy router inclusion

airflow-core/pyproject.toml

@@ -84,7 +84,7 @@ dependencies = [
     "asgiref>=2.3.0; python_version < '3.14'",
     "asgiref>=3.11.1; python_version >= '3.14'",
     "attrs>=22.1.0, !=25.2.0",
-    "cadwyn>=6.1.1",
+    "cadwyn>=7.1.0",
     "colorlog>=6.8.2",
     "cron-descriptor>=1.2.24",
     "croniter>=2.0.2",
@@ -95,11 +95,7 @@ dependencies = [
     "cryptography>=44.0.3",
     "deprecated>=1.2.13",
     "dill>=0.2.2",
-    # Cap below 0.137.0: FastAPI 0.137 switched to lazy router inclusion, which breaks cadwyn's
-    # versioned router generation (RouterGenerationError) and fails api-server / dag-processor
-    # startup. Relax once cadwyn supports FastAPI 0.137. See
-    # https://github.com/apache/airflow/issues/68562
-    "fastapi[standard-no-fastapi-cloud-cli]>=0.129.0,<0.137.0",
+    "fastapi[standard-no-fastapi-cloud-cli]>=0.137.2",
     "uvicorn>=0.37.0",
     "starlette>=1.0.1",
     "httpx>=0.25.0",

airflow-core/src/airflow/api_fastapi/execution_api/app.py

@@ -264,30 +264,9 @@ async def _extract_w3c_trace_context(
             otel_context.detach(token)
 
 
-def _inject_trace_context_dep(routes, mode: str) -> None:
-    dep = Depends(_extract_w3c_trace_context)
-    for route in routes:
-        if not isinstance(route, APIRoute):
-            continue
-        # Idempotent: create_task_execution_api_app() runs more than once per process
-        # (cached_app + InProcessExecutionAPI), and execution_api_router is shared
-        # module state, so strip any prior injection first.
-        route.dependencies[:] = [
-            d for d in route.dependencies if getattr(d, "dependency", None) is not _extract_w3c_trace_context
-        ]
-        match mode:
-            case "unsafe-always":
-                route.dependencies.insert(0, dep)
-            case "only-authenticated":
-                from airflow.api_fastapi.execution_api.security import require_auth
-
-                if any(getattr(d, "dependency", None) is require_auth for d in route.dependencies):
-                    route.dependencies.append(dep)
-
-
 def create_task_execution_api_app(lifespan: svcs.fastapi.lifespan = lifespan) -> FastAPI:
     """Create FastAPI app for task execution API."""
-    from airflow.api_fastapi.execution_api.routes import execution_api_router
+    from airflow.api_fastapi.execution_api.routes import build_execution_api_router
     from airflow.api_fastapi.execution_api.versions import bundle
     from airflow.configuration import conf
 
@@ -311,7 +290,11 @@ def custom_generate_unique_id(route: APIRoute):
     app.add_middleware(JWTReissueMiddleware)
 
     mode = conf.get("execution_api", "otel_trace_propagation", fallback="only-authenticated")
-    _inject_trace_context_dep(execution_api_router.routes, mode)
+    trace_context_dep = Depends(_extract_w3c_trace_context)
+    execution_api_router = build_execution_api_router(
+        pre_auth_dependencies=[trace_context_dep] if mode == "unsafe-always" else (),
+        post_auth_dependencies=[trace_context_dep] if mode == "only-authenticated" else (),
+    )
 
     app.generate_and_include_versioned_routers(execution_api_router)
 

airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py

@@ -16,8 +16,11 @@
 # under the License.
 from __future__ import annotations
 
+from collections.abc import Sequence
+
 from cadwyn import VersionedAPIRouter
 from fastapi import APIRouter, Security
+from fastapi.params import Depends as DependsParam
 
 from airflow.api_fastapi.execution_api.routes import (
     asset_events,
@@ -37,34 +40,46 @@
 )
 from airflow.api_fastapi.execution_api.security import require_auth
 
-execution_api_router = APIRouter()
-# health.router declares its full paths ("/health", "/health/ping") and is included without a
-# prefix, unlike the routers below. A root route registered as @router.get("") under an include-time
-# prefix=... raises "Prefix and path cannot be both empty" once FastAPI switched to lazy router
-# inclusion (>=0.137); see https://github.com/apache/airflow/issues/68562. Don't reintroduce a prefix here.
-execution_api_router.include_router(health.router, tags=["Health"])
 
-# _Every_ single endpoint under here must be authenticated. Some do further checks on top of these
-authenticated_router = VersionedAPIRouter(dependencies=[Security(require_auth)])  # type: ignore[list-item]
+def build_execution_api_router(
+    *,
+    pre_auth_dependencies: Sequence[DependsParam] = (),
+    post_auth_dependencies: Sequence[DependsParam] = (),
+) -> APIRouter:
+    """Assemble a fresh Task Execution API router tree with dependencies applied at build time."""
+    authenticated_router = VersionedAPIRouter(
+        dependencies=[Security(require_auth), *post_auth_dependencies]  # type: ignore[list-item]
+    )
 
-authenticated_router.include_router(assets.router, prefix="/assets", tags=["Assets"])
-authenticated_router.include_router(asset_events.router, prefix="/asset-events", tags=["Asset Events"])
-authenticated_router.include_router(
-    connection_tests.router, prefix="/connection-tests", tags=["Connection Tests"]
-)
-authenticated_router.include_router(connections.router, prefix="/connections", tags=["Connections"])
-authenticated_router.include_router(dag_runs.router, prefix="/dag-runs", tags=["Dag Runs"])
-authenticated_router.include_router(dags.router, prefix="/dags", tags=["Dags"])
-authenticated_router.include_router(task_instances.router, prefix="/task-instances", tags=["Task Instances"])
-authenticated_router.include_router(
-    task_reschedules.router, prefix="/task-reschedules", tags=["Task Reschedules"]
-)
-authenticated_router.include_router(variables.router, prefix="/variables", tags=["Variables"])
-authenticated_router.include_router(xcoms.router, prefix="/xcoms", tags=["XComs"])
-authenticated_router.include_router(hitl.router, prefix="/hitlDetails", tags=["Human in the Loop"])
-authenticated_router.include_router(task_state_store.router, prefix="/store/ti", tags=["Task State Store"])
-authenticated_router.include_router(
-    asset_state_store.router, prefix="/store/asset", tags=["Asset State Store"]
-)
+    authenticated_router.include_router(assets.router, prefix="/assets", tags=["Assets"])
+    authenticated_router.include_router(asset_events.router, prefix="/asset-events", tags=["Asset Events"])
+    authenticated_router.include_router(
+        connection_tests.router, prefix="/connection-tests", tags=["Connection Tests"]
+    )
+    authenticated_router.include_router(connections.router, prefix="/connections", tags=["Connections"])
+    authenticated_router.include_router(dag_runs.router, prefix="/dag-runs", tags=["Dag Runs"])
+    authenticated_router.include_router(dags.router, prefix="/dags", tags=["Dags"])
+    authenticated_router.include_router(
+        task_instances.router, prefix="/task-instances", tags=["Task Instances"]
+    )
+    authenticated_router.include_router(
+        task_reschedules.router, prefix="/task-reschedules", tags=["Task Reschedules"]
+    )
+    authenticated_router.include_router(variables.router, prefix="/variables", tags=["Variables"])
+    authenticated_router.include_router(xcoms.router, prefix="/xcoms", tags=["XComs"])
+    authenticated_router.include_router(hitl.router, prefix="/hitlDetails", tags=["Human in the Loop"])
+    authenticated_router.include_router(
+        task_state_store.router, prefix="/store/ti", tags=["Task State Store"]
+    )
+    authenticated_router.include_router(
+        asset_state_store.router, prefix="/store/asset", tags=["Asset State Store"]
+    )
 
-execution_api_router.include_router(authenticated_router)
+    execution_api_router = APIRouter()
+    # Health routes are intentionally outside authentication. FastAPI 0.137+ snapshots router
+    # dependencies during lazy inclusion, so both dependency layers must be supplied here.
+    execution_api_router.include_router(
+        health.router, tags=["Health"], dependencies=list(pre_auth_dependencies)
+    )
+    execution_api_router.include_router(authenticated_router, dependencies=list(pre_auth_dependencies))
+    return execution_api_router

airflow-core/tests/unit/api_fastapi/core_api/routes/test_routes.py

@@ -16,6 +16,9 @@
 # under the License.
 from __future__ import annotations
 
+from fastapi import APIRouter
+from fastapi.routing import iter_route_contexts
+
 from airflow.api_fastapi.core_api.routes.public import authenticated_router, public_router
 
 # Set of paths that are allowed to be accessible without authentication
@@ -27,21 +30,41 @@
 }
 
 
+def _get_paths_excluded_from_authentication(public: APIRouter, authenticated: APIRouter) -> set[str]:
+    public_paths = {route.path for route in iter_route_contexts(public.routes) if route.path is not None}
+    authenticated_paths = {
+        f"{public.prefix}{route.path}"
+        for route in iter_route_contexts(authenticated.routes)
+        if route.path is not None
+    }
+    return public_paths - authenticated_paths
+
+
 def test_no_auth_routes():
     """
     Verify that only the routes with NO_AUTH_PATHS are excluded from the `authenticated_router`. This
     test ensures that a router is not added to the non-authenticated router by mistake.
     """
-    paths_in_public_router = {route.path for route in public_router.routes}
-    paths_in_authenticated_router = {
-        f"{public_router.prefix}{route.path}" for route in authenticated_router.routes
+    assert _get_paths_excluded_from_authentication(public_router, authenticated_router) == NO_AUTH_PATHS
+
+
+def test_no_auth_routes_guard_detects_unprotected_route():
+    """Prove the guard reports a route accidentally registered outside the authenticated router."""
+    test_public_router = APIRouter(prefix="/api/v2")
+    test_authenticated_router = APIRouter()
+
+    @test_public_router.get("/unprotected")
+    def unprotected_route():
+        return None
+
+    assert _get_paths_excluded_from_authentication(test_public_router, test_authenticated_router) == {
+        "/api/v2/unprotected"
     }
-    assert paths_in_public_router - paths_in_authenticated_router == NO_AUTH_PATHS
 
 
 def test_routes_with_responses():
     """Verify that each route in `public_router` has appropriate responses configured."""
-    for route in public_router.routes:
+    for route in iter_route_contexts(public_router.routes):
         if route.path in NO_AUTH_PATHS:
             # Routes in NO_AUTH_PATHS should not have 401 or 403 response codes
             assert 401 not in route.responses, f"Route {route.path} should not require auth (401)"

airflow-core/tests/unit/api_fastapi/execution_api/test_app.py

@@ -233,20 +233,6 @@ def test_multiple_requests_with_different_correlation_ids(self, client):
 class TestTraceContextPropagation:
     """Exercise ``execution_api.otel_trace_propagation`` on the real Execution API app."""
 
-    @pytest.fixture(autouse=True)
-    def _restore_router_dependencies(self):
-        from airflow.api_fastapi.execution_api.routes import execution_api_router
-
-        snapshot = {
-            id(route): list(route.dependencies)
-            for route in execution_api_router.routes
-            if isinstance(route, APIRoute)
-        }
-        yield
-        for route in execution_api_router.routes:
-            if isinstance(route, APIRoute):
-                route.dependencies[:] = snapshot[id(route)]
-
     @staticmethod
     def _build_app(mode: str):
         with conf_vars({("execution_api", "otel_trace_propagation"): mode}):

airflow-core/tests/unit/api_fastapi/execution_api/test_token_scope_boundaries.py

@@ -37,9 +37,9 @@
 from __future__ import annotations
 
 import pytest
-from fastapi.routing import APIRoute
+from fastapi.routing import APIRoute, iter_route_contexts
 
-from airflow.api_fastapi.execution_api.routes import execution_api_router
+from airflow.api_fastapi.execution_api.routes import build_execution_api_router
 
 # Routes that intentionally deviate from the default (execution-only) policy.
 # Any route NOT listed here must accept only {"execution"}.
@@ -55,12 +55,13 @@
 def _all_route_policies() -> dict[str, set[str]]:
     """Return a map of all API routes and their allowed token types."""
     policy_map: dict[str, set[str]] = {}
-    for route in execution_api_router.routes:
-        if isinstance(route, APIRoute):
-            allowed_tokens = set(getattr(route, "allowed_token_types", {"execution"}))
-            if route.methods:
-                for method in route.methods:
-                    policy_map[f"{method} {route.path}"] = allowed_tokens
+    router = build_execution_api_router()
+    for route_context in iter_route_contexts(router.routes):
+        if isinstance(route_context.route, APIRoute):
+            allowed_tokens = set(getattr(route_context.original_route, "allowed_token_types", {"execution"}))
+            if route_context.methods:
+                for method in route_context.methods:
+                    policy_map[f"{method} {route_context.path}"] = allowed_tokens
     return policy_map
 
 

scripts/in_container/in_container_utils.py

@@ -20,15 +20,18 @@
 import subprocess
 import sys
 import textwrap
+from collections.abc import Sequence
 from contextlib import contextmanager
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Protocol, cast
 
 import rich_click as click
 from rich.console import Console
 
 if TYPE_CHECKING:
     from fastapi import FastAPI
+    from fastapi.routing import RouteContext
+    from starlette.routing import BaseRoute
 
 
 click.rich_click.COLOR_SYSTEM = "standard"
@@ -41,6 +44,10 @@
 AIRFLOW_DIST_PATH = Path("/dist")
 
 
+class _OpenAPIRoute(Protocol):
+    include_in_schema: bool
+
+
 @contextmanager
 def ci_group(group_name: str, github_actions: bool):
     if github_actions:
@@ -67,21 +74,29 @@ def run_command(cmd: list[str], github_actions: bool, **kwargs) -> subprocess.Co
 def generate_openapi_file(app: FastAPI, file_path: Path, prefix: str = "", only_ui: bool = False):
     import yaml
     from fastapi.openapi.utils import get_openapi
-    from fastapi.routing import APIRoute
+    from fastapi.routing import APIRoute, iter_route_contexts
 
+    routes: Sequence[BaseRoute | RouteContext] = app.routes
     if only_ui:
-        for route in app.routes:
-            if not isinstance(route, APIRoute):
-                continue
-            route.include_in_schema = route.path.startswith("/ui/")
+        route_contexts = [
+            route_context
+            for route_context in iter_route_contexts(app.routes)
+            if isinstance(route_context.original_route, APIRoute)
+            and (route_path := route_context.path) is not None
+            and route_path.startswith("/ui/")
+        ]
+        # UI routers are excluded from the public schema, so opt their materialized contexts back in.
+        for route_context in route_contexts:
+            cast("_OpenAPIRoute", route_context._effective_route).include_in_schema = True
+        routes = route_contexts
 
     with file_path.open("w+") as f:
         openapi_schema = get_openapi(
             title=app.title,
             version=app.version,
             openapi_version=app.openapi_version,
             description=app.description,
-            routes=app.routes,
+            routes=routes,
         )
         if prefix:
             openapi_schema["paths"] = {