Allow FastAPI 0.137+ by requiring cadwyn 8.1

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.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 @@
 
 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 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 @@
     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 custom_generate_unique_id(route: APIRoute):
         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 custom_generate_unique_id(route: APIRoute):
         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 pydantic import (
     AwareDatetime,
+    Discriminator,
     Field,
     JsonValue,
     Tag,
@@ -297,7 +298,7 @@ def ti_state_discriminator(v: dict[str, str] | StrictBaseModel) -> str:
     | 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 @@
     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 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 @@
 
 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 @@ async def api_fail():
     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.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 @@ def _dag_axis_methods(route) -> list[str]:
     @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 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)"