Restore FastAPI 0.137 support in the Task Execution API

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.1",
     "uvicorn>=0.37.0",
     "starlette>=1.0.1",
     "httpx>=0.25.0",

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

@@ -259,30 +259,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() -> 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
 
@@ -305,8 +284,16 @@ def custom_generate_unique_id(route: APIRoute):
     app.add_middleware(CorrelationIdMiddleware)
     app.add_middleware(JWTReissueMiddleware)
 
+    # FastAPI (>=0.137) freezes a router's dependencies into each route at include time, so the
+    # trace-context dependency has to be supplied while the router tree is assembled. "unsafe-always"
+    # extracts on every request (before auth); "only-authenticated" extracts only after a successful
+    # require_auth; "never" extracts nothing.
     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,6 +16,9 @@
 # under the License.
 from __future__ import annotations
 
+from collections.abc import Sequence
+from typing import Any
+
 from cadwyn import VersionedAPIRouter
 from fastapi import APIRouter, Security
 
@@ -37,34 +40,57 @@
 )
 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[Any] = (),
+    post_auth_dependencies: Sequence[Any] = (),
+) -> APIRouter:
+    """
+    Assemble the Task Execution API router tree.
 
-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"]
-)
+    ``pre_auth_dependencies`` run before authentication on every route (health included);
+    ``post_auth_dependencies`` run after ``require_auth`` and therefore only for successfully
+    authenticated requests. They must be supplied here, at build time: FastAPI (>=0.137) snapshots a
+    router's dependencies into each route when ``include_router`` is called, so dependencies attached
+    afterwards to a shared, already-assembled router never take effect.
+    """
+    # _Every_ single endpoint under here must be authenticated. Some do further checks on top of these.
+    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"]
+    )
 
-execution_api_router.include_router(authenticated_router)
+    execution_api_router = APIRouter()
+    # health.router declares its full paths ("/health", "/health/ping") and is included without a
+    # prefix, unlike the routers above. 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"], 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/execution_api/test_app.py

@@ -190,20 +190,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}):