Allow FastAPI 0.137+ by requiring cadwyn 8.1 (#75689)

* Allow FastAPI 0.137+ by requiring cadwyn 8.1

FastAPI was capped below 0.137 because its lazy router inclusion broke
cadwyn 7's versioned router generation. cadwyn 8.1 supports it, so lift
the cap and raise the cadwyn minimum. cadwyn 8.1 itself requires
fastapi>=0.137.1; raise our floor to 0.137.2, the first release with
fastapi.routing.iter_route_contexts, which the test helper relies on.

Fix what the newer FastAPI exposed: use pydantic's Discriminator for the
TIStateUpdate union, make Execution API trace-context injection reach
routes inside included routers (it was silently a no-op), and update
tests that relied on the removed get_flat_dependant or on flattened
router.routes.

related: #68562

* Stop FastAPI's native telemetry and lazy routers changing API behavior

FastAPI 0.142 added native OpenTelemetry that is on by default and runs
before routing and auth. With a tracer provider configured it extracted
caller trace context on unauthenticated requests and exported query
strings unredacted, bypassing [execution_api] otel_trace_propagation.
Exclude every request on the apps Airflow builds, which also stops the
apps mounted under them from tracing, and raise the FastAPI floor to
0.142.0, the first release with that option.

Lazy router inclusion also broke code reading app.routes: HTTP metric
route tags lost their router prefixes, the private UI OpenAPI spec
regenerated without its /ui routes, and the edge3 worker API route
tests saw no routes.

related: #68562

* Detect Execution API auth from enclosing routers and drop cadwyn changelog

Trace-context extraction in only-authenticated mode now checks each route's
merged dependencies, so authentication added by an enclosing router or an
include_router() call still counts. Walking routes through FastAPI's public
iter_route_contexts also avoids relying on its private included-router type.

Cadwyn 8.1 deprecates the changelog endpoint and warns whenever it is enabled;
the Execution API never used it.

airflow-core/pyproject.toml

@@ -81,7 +81,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>=8.1.0",
     "colorlog>=6.8.2",
     "cron-descriptor>=1.2.24",
     "croniter>=6.2.2",
@@ -92,11 +92,8 @@ 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",
+    # 0.142.0 is the first release with the ``telemetry`` option used to turn off FastAPI's native OpenTelemetry.
+    "fastapi[standard-no-fastapi-cloud-cli]>=0.142.0",
     "uvicorn>=0.37.0",
     # 1.3.1 is the first release whose urlencoded FormParser enforces max_fields/max_part_size
     # and reports them as a 400. Below it, an unauthenticated form body to POST /auth/token is

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

@@ -32,6 +32,7 @@ from airflow.api_fastapi.common.dagbag import create_dag_bag
 from airflow.api_fastapi.common.exceptions import init_error_handlers
 from airflow.api_fastapi.common.http_access_log import HttpAccessLogMiddleware
 from airflow.api_fastapi.common.http_metrics import HttpMetricsMiddleware
+from airflow.api_fastapi.common.telemetry import NATIVE_TELEMETRY_DISABLED
 from airflow.api_fastapi.core_api.app import (
     init_config,
     init_flask_plugins,
@@ -154,6 +155,7 @@ def create_app(apps: str = "all") -> FastAPI:
         "supported by these parameters; regex-capable endpoints expose a separate parameter.",
         lifespan=lifespan,
         root_path=API_ROOT_PATH.removesuffix("/"),
+        telemetry=NATIVE_TELEMETRY_DISABLED,
         version="2",
         docs_url="/docs" if conf.getboolean("api", "enable_swagger_ui") else None,
         redoc_url="/redoc" if conf.getboolean("api", "enable_swagger_ui") else None,

airflow-core/src/airflow/api_fastapi/common/http_metrics.py

@@ -19,9 +19,11 @@
 from __future__ import annotations
 
 import time
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
+from weakref import WeakKeyDictionary
 
 import structlog
+from fastapi.routing import iter_route_contexts
 
 from airflow._shared.observability.metrics.stats import Stats
 
@@ -30,6 +32,8 @@ if TYPE_CHECKING:
 
 logger = structlog.get_logger(logger_name="http.metrics")
 
+_ROUTE_PATHS: WeakKeyDictionary[Any, dict[int, str]] = WeakKeyDictionary()
+
 # `/execution` is the Task SDK execution API, mounted as a sub-app. The middleware runs on the
 # outer app, so a request to it passes through here like any other.
 _API_METRICS_PATH_PREFIXES = ("/api/v2", "/ui", "/execution")
@@ -51,11 +55,28 @@ def _get_method_tag(method: str) -> str:
     return method if method in _SERVED_HTTP_METHODS else _OTHER_HTTP_METHOD
 
 
+def _get_route_paths(app: Any) -> dict[int, str]:
+    """Return the full path of each route in ``app``, keyed by ``id()`` of the route object."""
+    paths = _ROUTE_PATHS.get(app)
+    if paths is None:
+        paths = {}
+        for context in iter_route_contexts(app.routes):
+            if context.path:
+                paths.setdefault(id(context.original_route), context.path)
+        _ROUTE_PATHS[app] = paths
+    return paths
+
+
 def _get_route_template(scope: Scope) -> str:
     # FastAPI's APIRoute.matches() stores the matched route in the scope before the endpoint runs.
     # Requests matching no route (404s, redirects) share one bucket so tag cardinality stays bounded.
-    route_path = getattr(scope.get("route"), "path", None)
+    route = scope.get("route")
+    route_path = getattr(route, "path", None)
     if isinstance(route_path, str) and route_path:
+        # The route in the scope is the one declared on its own router, so its path lacks the
+        # prefixes of the routers it is included through (`/dags` rather than `/api/v2/dags`).
+        if (app := scope.get("app")) is not None:
+            route_path = _get_route_paths(app).get(id(route), route_path)
         # A mounted sub-app declares its routes relative to the mount point, which Starlette keeps
         # in root_path. Without it, `/execution/task-instances/{id}` would be tagged
         # `/task-instances/{id}` and collide with core routes sharing that template.

airflow-core/src/airflow/api_fastapi/common/telemetry.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.
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import MutableMapping
+
+    from fastapi.telemetry import TelemetryConfig
+
+
+def _exclude_every_request(scope: MutableMapping[str, Any]) -> bool:
+    return True
+
+
+# Airflow controls request tracing itself (``[execution_api] otel_trace_propagation``, query
+# string redaction), so FastAPI's native OpenTelemetry must stay out of the way. Excluding every
+# request, rather than turning the signals off, also keeps the FastAPI apps mounted underneath
+# (auth managers, plugins) from tracing it.
+NATIVE_TELEMETRY_DISABLED: TelemetryConfig = {
+    "exclude": _exclude_every_request,
+    "auto_configure": False,
+}

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

@@ -30,7 +30,7 @@ from cadwyn import (
 )
 from fastapi import Depends, FastAPI, Request, Response
 from fastapi.responses import JSONResponse
-from fastapi.routing import APIRoute
+from fastapi.routing import APIRoute, iter_route_contexts
 from opentelemetry import context as otel_context, propagate as otel_propagate
 from starlette.middleware.base import BaseHTTPMiddleware
 
@@ -40,8 +40,9 @@ from airflow.api_fastapi.auth.tokens import (
     get_sig_validation_args,
     get_signing_args,
 )
+from airflow.api_fastapi.common.telemetry import NATIVE_TELEMETRY_DISABLED
 from airflow.api_fastapi.execution_api.in_process import InProcessExecutionAPI
-from airflow.api_fastapi.execution_api.security import _REQUEST_SCOPE_TOKEN_KEY
+from airflow.api_fastapi.execution_api.security import _REQUEST_SCOPE_TOKEN_KEY, require_auth
 
 if TYPE_CHECKING:
     from airflow.models.dagbag import DBDagBag
@@ -265,25 +266,30 @@ async def _extract_w3c_trace_context(
             otel_context.detach(token)
 
 
+def _is_trace_context_dep(dependency) -> bool:
+    return getattr(dependency, "dependency", None) is _extract_w3c_trace_context
+
+
 def _inject_trace_context_dep(routes, mode: str) -> None:
-    dep = Depends(_extract_w3c_trace_context)
-    for route in routes:
+    """
+    Add trace-context extraction to the authenticated routes in ``only-authenticated`` mode.
+
+    ``unsafe-always`` mode is handled with an app-level dependency instead, so extraction runs
+    before authentication declared on any enclosing router.
+    """
+    for context in iter_route_contexts(routes):
+        route = context.original_route
         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)
+        route.dependencies[:] = [d for d in route.dependencies if not _is_trace_context_dep(d)]
+        if mode != "only-authenticated":
+            continue
+        # context.dependencies also holds those of the enclosing routers and include_router() calls.
+        if any(getattr(d, "dependency", None) is require_auth for d in context.dependencies):
+            route.dependencies.append(Depends(_extract_w3c_trace_context))
 
 
 def create_task_execution_api_app(
@@ -308,6 +314,7 @@ def create_task_execution_api_app(
         return route.name
 
     # See https://docs.cadwyn.dev/concepts/version_changes/ for info about API versions
+    mode = conf.get("execution_api", "otel_trace_propagation", fallback="only-authenticated")
     app = CadwynWithOpenAPICustomization(
         title="Airflow Task Execution API",
         description="The private Airflow Task Execution API.",
@@ -316,13 +323,15 @@ def create_task_execution_api_app(
         api_version_parameter_name="Airflow-API-Version",
         api_version_default_value=bundle.versions[0].value,
         versions=bundle,
+        changelog_url=None,
+        dependencies=[Depends(_extract_w3c_trace_context)] if mode == "unsafe-always" else None,
+        telemetry=NATIVE_TELEMETRY_DISABLED,
     )
 
     # Add correlation-id middleware for request tracing
     app.add_middleware(CorrelationIdMiddleware)
     app.add_middleware(JWTReissueMiddleware)
 
-    mode = conf.get("execution_api", "otel_trace_propagation", fallback="only-authenticated")
     _inject_trace_context_dep(execution_api_router.routes, mode)
 
     app.generate_and_include_versioned_routers(execution_api_router)

airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py

@@ -24,6 +24,7 @@ from typing import TYPE_CHECKING, Annotated, Any, Literal, TypeVar
 
 from pydantic import (
     AwareDatetime,
+    Discriminator,
     Field,
     JsonValue,
     Tag,
@@ -297,7 +298,7 @@ TIStateUpdate = Annotated[
     | Annotated[TIRescheduleStatePayload, Tag("up_for_reschedule")]
     | Annotated[TIAwaitingInputStatePayload, Tag("awaiting_input")]
     | Annotated[TIRetryStatePayload, Tag("up_for_retry")],
-    Field(discriminator=ti_state_discriminator),
+    Discriminator(ti_state_discriminator),
 ]
 
 

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

@@ -78,11 +78,13 @@ Why ``ExecutionAPIRoute`` is needed:
     to *relax* the outer router's default execution-only restriction, but
     ``SecurityScopes`` only accumulate additively — an outer dependency
     cannot see scopes declared by inner ones. ``ExecutionAPIRoute`` solves
-    this by inspecting the **merged** dependency list at route registration
-    time (after ``include_router`` has combined all parent and child
-    dependencies) and precomputing the full ``allowed_token_types`` set.
+    this by inspecting the route's own dependency list at route registration
+    time and precomputing the full ``allowed_token_types`` set.
     ``require_auth`` then reads this precomputed set from the matched route
-    at request time, avoiding the ordering problem entirely.
+    at request time, avoiding the ordering problem entirely. FastAPI includes
+    routers lazily, so ``token:*`` scopes must be declared on the route
+    itself: scopes on an enclosing router or ``include_router`` call are not
+    seen, and the route falls back to execution-only.
 
     Any router whose routes need non-default token type policies must use
     ``route_class=ExecutionAPIRoute``. Routers that only need the default

airflow-core/src/airflow/utils/serve_logs/log_server.py

@@ -35,6 +35,7 @@ from jwt.exceptions import (
 
 from airflow._shared.module_loading import import_string
 from airflow.api_fastapi.auth.tokens import JWTValidator, get_signing_key
+from airflow.api_fastapi.common.telemetry import NATIVE_TELEMETRY_DISABLED
 from airflow.configuration import conf
 from airflow.utils.docs import get_docs_url
 
@@ -140,7 +141,7 @@ def create_app():
         except Exception as e:
             raise ImportError(f"Unable to load {log_config_class} due to error: {e}")
 
-    fastapi_app = FastAPI()
+    fastapi_app = FastAPI(telemetry=NATIVE_TELEMETRY_DISABLED)
     fastapi_app.state.signer = JWTValidator(
         issuer=None,
         secret_key=get_signing_key("api", "secret_key"),

airflow-core/tests/unit/api_fastapi/common/test_http_metrics.py

@@ -21,7 +21,7 @@ from unittest import mock
 
 import pytest
 import structlog.testing
-from fastapi import FastAPI
+from fastapi import APIRouter, FastAPI
 from starlette.responses import PlainTextResponse
 from starlette.testclient import TestClient
 
@@ -76,6 +76,16 @@ def test_app() -> FastAPI:
     async def health():
         return PlainTextResponse("healthy")
 
+    dags_router = APIRouter(prefix="/dags")
+
+    @dags_router.get("/{dag_id}")
+    async def included_dag(dag_id: str):
+        return PlainTextResponse("ok")
+
+    public_router = APIRouter(prefix="/api/v2")
+    public_router.include_router(dags_router)
+    app.include_router(public_router)
+
     app.mount("/execution", _make_execution_app())
     app.add_middleware(HttpMetricsMiddleware)
     return app
@@ -160,6 +170,7 @@ def test_request_duration_is_emitted_in_milliseconds():
             id="route-template-with-parameters",
         ),
         pytest.param("/ui/items/42", "/ui/items/{item_id}", id="ui"),
+        pytest.param("/api/v2/dags/example", "/api/v2/dags/{dag_id}", id="included-router-keeps-prefixes"),
         pytest.param(
             "/execution/task-instances/abc/run",
             "/execution/task-instances/{task_instance_id}/run",

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py

@@ -59,6 +59,7 @@ from airflow.utils.session import provide_session
 from airflow.utils.state import DagRunState, DagSchedulingState
 from airflow.utils.types import DagRunType
 
+from tests_common.test_utils.api_fastapi import iter_api_routes
 from tests_common.test_utils.asserts import assert_queries_count
 from tests_common.test_utils.config import conf_vars
 from tests_common.test_utils.db import (
@@ -2062,7 +2063,9 @@ class TestQueuedEventsDagAxisAuthorization:
     @pytest.fixture
     def routes_by_path(self, test_client):
         return {
-            (r.path, tuple(sorted(r.methods))): r for r in test_client.app.routes if hasattr(r, "dependant")
+            (r.path, tuple(sorted(r.methods))): r
+            for r in iter_api_routes(test_client.app.routes)
+            if hasattr(r, "dependant")
         }
 
     @pytest.mark.parametrize(

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

@@ -21,6 +21,8 @@ from fastapi.openapi.utils import get_openapi
 
 from airflow.api_fastapi.core_api.routes.public import authenticated_router, public_router
 
+from tests_common.test_utils.api_fastapi import iter_api_routes
+
 # Set of paths that are allowed to be accessible without authentication
 NO_AUTH_PATHS = {
     "/api/v2/auth/login",
@@ -35,16 +37,16 @@ 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_public_router = {route.path for route in iter_api_routes(public_router.routes)}
     paths_in_authenticated_router = {
-        f"{public_router.prefix}{route.path}" for route in authenticated_router.routes
+        f"{public_router.prefix}{route.path}" for route in iter_api_routes(authenticated_router.routes)
     }
     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_api_routes(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

@@ -25,7 +25,7 @@ from uuid import UUID
 
 import httpx2
 import pytest
-from fastapi import FastAPI, Request, status
+from fastapi import APIRouter, Depends, FastAPI, Request, status
 from fastapi.params import Security as SecurityParam
 from fastapi.routing import APIRoute
 from fastapi.testclient import TestClient
@@ -38,6 +38,8 @@ from airflow.api_fastapi.app import create_app
 from airflow.api_fastapi.execution_api.app import (
     InProcessExecutionAPI,
     _extract_w3c_trace_context,
+    _inject_trace_context_dep,
+    _is_trace_context_dep,
     create_jwt_generator,
     create_task_execution_api_app,
 )
@@ -48,6 +50,7 @@ from airflow.api_fastapi.execution_api.versions import bundle
 from airflow.models.xcom import XComModel
 from airflow.utils.session import create_session_async
 
+from tests_common.test_utils.api_fastapi import iter_api_routes
 from tests_common.test_utils.config import conf_vars
 
 pytestmark = pytest.mark.db_test
@@ -397,18 +400,14 @@ class TestTraceContextPropagation:
     """Exercise ``execution_api.otel_trace_propagation`` on the real Execution API app."""
 
     @pytest.fixture(autouse=True)
-    def _restore_router_dependencies(self):
+    def _restore_route_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)
-        }
+        routes = [context.original_route for context in iter_api_routes(execution_api_router.routes)]
+        snapshot = [list(route.dependencies) for route in routes]
         yield
-        for route in execution_api_router.routes:
-            if isinstance(route, APIRoute):
-                route.dependencies[:] = snapshot[id(route)]
+        for route, dependencies in zip(routes, snapshot):
+            route.dependencies[:] = dependencies
 
     @staticmethod
     def _build_app(mode: str):
@@ -454,6 +453,16 @@ class TestTraceContextPropagation:
         if expect_status is not None:
             assert response.status_code == expect_status
 
+    def test_only_authenticated_detects_auth_added_by_include_router(self):
+        router = APIRouter()
+        router.add_api_route("/variables/{key}", lambda key: None)
+        parent = APIRouter()
+        parent.include_router(router, dependencies=[Depends(require_auth)])
+
+        _inject_trace_context_dep(parent.routes, "only-authenticated")
+
+        assert any(_is_trace_context_dep(d) for d in router.routes[0].dependencies)
+
     def test_trace_context_dep_cleans_up_on_route_exception(self):
         """Verify extract and cleanup run correctly when a route handler raises."""
         app = self._build_app("unsafe-always")

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

@@ -40,13 +40,14 @@ from typing import get_args
 
 import pytest
 from fastapi.dependencies.models import Dependant
-from fastapi.dependencies.utils import get_flat_dependant
 from fastapi.routing import APIRoute
 
 from airflow.api_fastapi.execution_api.routes import execution_api_router
 from airflow.api_fastapi.execution_api.security import get_selected_dag_bundle, require_dag_in_granted_bundle
 from airflow.dag_processing.processor import ToManager
 
+from tests_common.test_utils.api_fastapi import iter_api_routes
+
 # Routes that intentionally deviate from the default (execution-only) policy.
 # Any route NOT listed here must accept only {"execution"}.
 NON_DEFAULT_TOKEN_POLICY: dict[str, set[str]] = {
@@ -113,11 +114,15 @@ DAG_PROCESSOR_MESSAGE_ROUTES: dict[str, str | None] = {
 }
 
 
+def _get_original_route(route) -> APIRoute:
+    """Return the ``APIRoute`` itself; custom attributes are not forwarded by FastAPI's route contexts."""
+    return getattr(route, "original_route", route)
+
+
 def _get_api_routes() -> dict[str, APIRoute]:
     return {
         f"{method} {route.path}": route
-        for route in execution_api_router.routes
-        if isinstance(route, APIRoute)
+        for route in iter_api_routes(execution_api_router.routes)
         for method in route.methods or ()
     }
 
@@ -130,10 +135,17 @@ def _get_dependency_calls(dependant: Dependant) -> set:
     return calls
 
 
+def _get_path_and_query_param_names(dependant: Dependant) -> set[str]:
+    names = {param.name for param in (*dependant.path_params, *dependant.query_params)}
+    for dependency in dependant.dependencies:
+        names |= _get_path_and_query_param_names(dependency)
+    return names
+
+
 def _all_route_policies() -> dict[str, set[str]]:
     """Return a map of all API routes and their allowed token types."""
     return {
-        key: set(getattr(route, "allowed_token_types", {"execution"}))
+        key: set(getattr(_get_original_route(route), "allowed_token_types", {"execution"}))
         for key, route in _get_api_routes().items()
     }
 
@@ -198,7 +210,9 @@ class TestDagProcessorMessageRoutes:
 
     def test_only_job_completion_skips_the_open_job_check(self):
         skipping = {
-            key for key, route in _get_api_routes().items() if not getattr(route, "requires_open_job", True)
+            key
+            for key, route in _get_api_routes().items()
+            if not getattr(_get_original_route(route), "requires_open_job", True)
         }
 
         assert skipping == JOB_UNCHECKED_ROUTES
@@ -206,8 +220,7 @@ class TestDagProcessorMessageRoutes:
     @pytest.mark.parametrize("route_key", sorted(filter(None, DAG_PROCESSOR_MESSAGE_ROUTES.values())))
     def test_dag_processor_route_is_bound_to_a_granted_bundle(self, route_key):
         route = _get_api_routes()[route_key]
-        flat = get_flat_dependant(route.dependant)
-        takes_dag_id = any(param.name == "dag_id" for param in (*flat.path_params, *flat.query_params))
+        takes_dag_id = "dag_id" in _get_path_and_query_param_names(route.dependant)
         expected = require_dag_in_granted_bundle if takes_dag_id else get_selected_dag_bundle
 
         assert expected in _get_dependency_calls(route.dependant)

airflow-core/tests/unit/api_fastapi/logging/test_decorators.py

@@ -23,8 +23,7 @@ from unittest.mock import MagicMock
 
 import pytest
 from fastapi import Request
-from fastapi.dependencies.utils import get_flat_dependant
-from fastapi.routing import APIRoute
+from fastapi.dependencies.models import Dependant
 from sqlalchemy import delete, func, select
 from sqlalchemy.orm import Session
 
@@ -42,6 +41,7 @@ from airflow.models.dagbundle import DagBundleModel
 from airflow.models.team import Team
 from airflow.providers.standard.operators.empty import EmptyOperator
 
+from tests_common.test_utils.api_fastapi import iter_api_routes
 from tests_common.test_utils.config import conf_vars
 from tests_common.test_utils.db import (
     clear_db_connections,
@@ -561,27 +561,31 @@ class TestActionLoggingUnparsableBody:
         assert logged.event == "test_event"
 
 
+def _has_body_params(dependant: Dependant) -> bool:
+    """Return True if the dependant or any of its sub-dependencies declares a body parameter."""
+    return bool(dependant.body_params) or any(_has_body_params(dep) for dep in dependant.dependencies)
+
+
 def _bodyless_action_logging_routes(app) -> list[tuple[str, str]]:
     """Every ``action_logging`` route that declares no request body of its own."""
     found = set()
 
     def _collect(route):
-        if not isinstance(route, APIRoute):
-            return
         uses_logging = any(
             getattr(dep.call, "__qualname__", "").startswith("action_logging")
             for dep in route.dependant.dependencies
         )
         if not uses_logging:
             return
-        if get_flat_dependant(route.dependant, skip_repeats=True).body_params:
+        if _has_body_params(route.dependant):
             return
         for method in route.methods or []:
             found.add((method, route.path))
 
-    for route in app.routes:
+    for route in iter_api_routes(app.routes):
         _collect(route)
-        for sub in getattr(getattr(route, "app", None), "routes", []) or []:
+    for mount in app.routes:
+        for sub in iter_api_routes(getattr(getattr(mount, "app", None), "routes", []) or []):
             _collect(sub)
     return sorted(found)
 

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

@@ -24,6 +24,9 @@ from unittest import mock
 import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
 from sqlalchemy import event, text
 from sqlalchemy.engine import Engine
 from starlette.middleware.gzip import GZipMiddleware
@@ -213,6 +216,28 @@ def test_api_metrics_middleware_skipped_without_a_metrics_backend(apps):
     assert HttpMetricsMiddleware not in installed
 
 
+@pytest.mark.parametrize(
+    "path",
+    [
+        pytest.param("/api/v2/monitor/health", id="root-app"),
+        pytest.param("/auth/token", id="mounted-auth-manager-app"),
+        pytest.param("/execution/health", id="mounted-execution-app"),
+    ],
+)
+def test_fastapi_native_telemetry_records_no_spans(path):
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    traceparent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
+
+    with mock.patch("opentelemetry.trace.get_tracer_provider", return_value=provider):
+        TestClient(app_module.create_app(apps="all")).get(path, headers={"traceparent": traceparent})
+
+    assert [
+        span.name for span in exporter.get_finished_spans() if span.instrumentation_scope.name == "fastapi"
+    ] == []
+
+
 def test_jwt_refresh_middleware_skips_static_mount():
     """Static UI assets are served without resolving the user from the JWT cookie."""
     [middleware] = [

devel-common/src/tests_common/test_utils/api_fastapi.py

@@ -17,6 +17,8 @@
 from __future__ import annotations
 
 import json
+from collections.abc import Iterable, Iterator
+from typing import Any
 from uuid import UUID
 
 from airflow.models import DagRun, Log
@@ -102,3 +104,24 @@ def _check_task_instance_note(session, ti_id, note_data):
         session.refresh(ti_note)
         assert ti_note.content == note_data["content"]
         assert ti_note.user_id == note_data["user_id"]
+
+
+def iter_api_routes(routes: Iterable[Any]) -> Iterator[Any]:
+    """
+    Yield every ``APIRoute`` of a router or app with its full path, walking included routers.
+
+    FastAPI 0.137+ includes routers lazily, so this yields ``RouteContext`` objects; set custom
+    attributes on ``context.original_route``. Older FastAPI, still used by provider compatibility
+    tests, flattens included routers into ``routes``, so the ``APIRoute`` objects are yielded as is.
+    """
+    from fastapi.routing import APIRoute
+
+    try:
+        from fastapi.routing import iter_route_contexts
+    except ImportError:
+        yield from (route for route in routes if isinstance(route, APIRoute))
+        return
+
+    for context in iter_route_contexts(list(routes)):
+        if isinstance(context.original_route, APIRoute):
+            yield context

providers/edge3/tests/unit/edge3/worker_api/test_app.py

@@ -20,11 +20,12 @@ import mimetypes
 from unittest import mock
 
 import pytest
-from fastapi.routing import APIRoute
 from starlette.routing import Mount
 
 from airflow.providers.edge3.worker_api.app import create_edge_worker_api_app
 
+from tests_common.test_utils.api_fastapi import iter_api_routes
+
 
 @pytest.fixture
 def app():
@@ -37,7 +38,7 @@ def app():
 
 
 def _api_route_paths(app) -> set[str]:
-    return {route.path for route in app.routes if isinstance(route, APIRoute)}
+    return {route.path for route in iter_api_routes(app.routes)}
 
 
 class TestCreateEdgeWorkerApiApp:

scripts/in_container/in_container_utils.py

@@ -29,6 +29,8 @@ 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"
@@ -67,13 +69,17 @@ 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 iter_route_contexts
 
+    routes: list[BaseRoute | RouteContext] = list(app.routes)
     if only_ui:
-        for route in app.routes:
-            if not isinstance(route, APIRoute):
-                continue
-            route.include_in_schema = route.path.startswith("/ui/")
+        routes = []
+        for context in iter_route_contexts(app.routes):
+            if context.path and context.path.startswith("/ui/"):
+                # ui_router sets include_in_schema=False to keep these routes out of the public spec, and
+                # FastAPI fixes that flag on an included route when it first resolves it.
+                object.__setattr__(context, "include_in_schema", True)
+                routes.append(context)
 
     with file_path.open("w+") as f:
         openapi_schema = get_openapi(
@@ -81,7 +87,7 @@ def generate_openapi_file(app: FastAPI, file_path: Path, prefix: str = "", only_
             version=app.version,
             openapi_version=app.openapi_version,
             description=app.description,
-            routes=app.routes,
+            routes=routes,
         )
         if prefix:
             openapi_schema["paths"] = {

task-sdk/pyproject.toml

@@ -52,7 +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",
+    "cadwyn>=8.1.0",
     "fsspec>=2023.10.0",
     "greenback>=1.2.1",
     "httpx2>=2.0.0",

uv.lock

@@ -2137,7 +2137,7 @@ requires-dist = [
     { name = "asgiref", marker = "python_full_version >= '3.14'", specifier = ">=3.11.1" },
     { name = "attrs", specifier = ">=22.1.0,!=25.2.0" },
     { name = "cachetools", specifier = ">=6.0.0" },
-    { name = "cadwyn", specifier = ">=6.1.1" },
+    { name = "cadwyn", specifier = ">=8.1.0" },
     { name = "colorlog", specifier = ">=6.8.2" },
     { name = "cron-descriptor", specifier = ">=1.2.24" },
     { name = "croniter", specifier = ">=6.2.2" },
@@ -2145,7 +2145,7 @@ requires-dist = [
     { name = "deprecated", specifier = ">=1.2.13" },
     { name = "dill", specifier = ">=0.2.2" },
     { name = "eventlet", marker = "extra == 'async'", specifier = ">=0.37.0" },
-    { name = "fastapi", extras = ["standard-no-fastapi-cloud-cli"], specifier = ">=0.129.0,<0.137.0" },
+    { name = "fastapi", extras = ["standard-no-fastapi-cloud-cli"], specifier = ">=0.142.0" },
     { name = "gevent", marker = "extra == 'async'", specifier = ">=25.4.1" },
     { name = "graphviz", marker = "sys_platform != 'darwin' and extra == 'graphviz'", specifier = ">=0.20" },
     { name = "greenback", marker = "extra == 'async'", specifier = ">=1.2.1" },
@@ -9503,7 +9503,7 @@ requires-dist = [
     { name = "asgiref", marker = "python_full_version >= '3.14'", specifier = ">=3.11.1" },
     { name = "attrs", specifier = ">=24.2.0,!=25.2.0" },
     { name = "babel", specifier = ">=2.17.0" },
-    { name = "cadwyn", specifier = ">=6.1.1" },
+    { name = "cadwyn", specifier = ">=8.1.0" },
     { name = "colorlog", specifier = ">=6.8.2" },
     { name = "datadog", marker = "extra == 'all'", specifier = ">=0.50.0" },
     { name = "datadog", marker = "extra == 'datadog'", specifier = ">=0.50.0" },
@@ -11032,7 +11032,7 @@ wheels = [
 
 [[package]]
 name = "cadwyn"
-version = "7.0.0"
+version = "8.1.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "fastapi" },
@@ -11042,9 +11042,9 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "typing-inspection" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/80/31/9a986a9fe20c6b52bde0dd23c9fc002388ab0c8f7b30a37217b07aa1875f/cadwyn-7.0.0.tar.gz", hash = "sha256:3b57549a37e218dffb55ac5d188639de0516207f05642b084f23627c5a44d614", size = 662353, upload-time = "2026-06-06T16:34:39.492Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f7/68/1f66b4f862aa678d41d7d84f16b0058b10fe65cf6d79d42693c3c9d1d64e/cadwyn-8.1.0.tar.gz", hash = "sha256:f5fab5014c9375550844e09c0bd010108d0423972a096f0d7151d74f891ceeec", size = 909665, upload-time = "2026-10-22T06:42:59.275Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/73/c4/efee5781dc8b3a50fd876d413cad6db6ee69e1f21d5a07ca469cec03cd22/cadwyn-7.0.0-py3-none-any.whl", hash = "sha256:727d3c444ae992bb2a238246d13f173e4802c92e1c901458975549e6f0522560", size = 61194, upload-time = "2026-06-06T16:34:37.768Z" },
+    { url = "https://files.pythonhosted.org/packages/23/52/c3af911a71c38e596f2f5f6c91fa9cdd92f595a0e8ded71b731a4347a675/cadwyn-8.1.0-py3-none-any.whl", hash = "sha256:6a90765af1ce1b6ca677b358f0987ed1626495162ee0d53c82ddc32ef8d7f6fa", size = 120961, upload-time = "2026-10-22T06:43:00.384Z" },
 ]
 
 [[package]]
@@ -12675,18 +12675,19 @@ wheels = [
 
 [[package]]
 name = "fastapi"
-version = "0.136.3"
+version = "0.161.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "annotated-doc" },
+    { name = "opentelemetry-api" },
     { name = "pydantic" },
     { name = "starlette" },
     { name = "typing-extensions" },
     { name = "typing-inspection" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/81/2d/ff8d91d7b564d464629a0fd50a4489c97fcb836ac230bf3a7269232a9b1f/fastapi-0.136.3.tar.gz", hash = "sha256:e487fae93ad408e6f47641ee4dfe389864fd7bec92e547ea8498fc13f43e83ab", size = 396410, upload-time = "2026-05-23T18:53:15.192Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/ee/12/f5a074dd4efd40898b1270551fdba19202f17591895dbc8429755128bd1b/fastapi-0.161.0.tar.gz", hash = "sha256:c65a5226b55061349a5d5f4940a1c45da6c08ec40e58211aed9e3e1baa4e7ba5", size = 527298, upload-time = "2026-10-22T18:30:43.421Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/e0/82/45359b62a067409bd929ae8a56b8ed13e5a8c8a61194b3c236920999ab83/fastapi-0.136.3-py3-none-any.whl", hash = "sha256:3d2a69bdf04b7e9f3afa292c3bc7a98816bbfafa10bc9b45f3f3700d2f761620", size = 117481, upload-time = "2026-05-23T18:53:16.924Z" },
+    { url = "https://files.pythonhosted.org/packages/f3/48/3cd6f8f30857fef456c03ae1ef89c1a82aba8a3a229896c28967b49849e1/fastapi-0.161.0-py3-none-any.whl", hash = "sha256:843d71746b18a3349e2c0b56993631788e6d89381c1973d7d8f4aa098f623ecb", size = 149008, upload-time = "2026-10-22T18:30:41.955Z" },
 ]
 
 [package.optional-dependencies]
@@ -12694,7 +12695,10 @@ standard-no-fastapi-cloud-cli = [
     { name = "email-validator" },
     { name = "fastapi-cli", extra = ["standard-no-fastapi-cloud-cli"] },
     { name = "httpx" },
+    { name = "httpx2" },
     { name = "jinja2" },
+    { name = "opentelemetry-exporter-otlp-proto-http" },
+    { name = "opentelemetry-sdk" },
     { name = "pydantic-extra-types" },
     { name = "pydantic-settings" },
     { name = "python-multipart" },