This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new b70a439a238 Allow FastAPI 0.137+ by requiring cadwyn 8.1 (#75689)
b70a439a238 is described below
commit b70a439a23827a140ed75498f39c429089f723b5
Author: Jed Cunningham <[email protected]>
AuthorDate: Mon Oct 26 22:36:04 2026 -0600
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 | 9 ++---
airflow-core/src/airflow/api_fastapi/app.py | 2 ++
.../src/airflow/api_fastapi/common/http_metrics.py | 25 +++++++++++--
.../src/airflow/api_fastapi/common/telemetry.py | 38 ++++++++++++++++++++
.../src/airflow/api_fastapi/execution_api/app.py | 41 +++++++++++++---------
.../execution_api/datamodels/taskinstance.py | 3 +-
.../airflow/api_fastapi/execution_api/security.py | 10 +++---
.../src/airflow/utils/serve_logs/log_server.py | 3 +-
.../unit/api_fastapi/common/test_http_metrics.py | 13 ++++++-
.../core_api/routes/public/test_assets.py | 5 ++-
.../api_fastapi/core_api/routes/test_routes.py | 8 +++--
.../unit/api_fastapi/execution_api/test_app.py | 29 +++++++++------
.../execution_api/test_token_scope_boundaries.py | 27 ++++++++++----
.../unit/api_fastapi/logging/test_decorators.py | 18 ++++++----
airflow-core/tests/unit/api_fastapi/test_app.py | 25 +++++++++++++
.../src/tests_common/test_utils/api_fastapi.py | 23 ++++++++++++
.../edge3/tests/unit/edge3/worker_api/test_app.py | 5 +--
scripts/in_container/in_container_utils.py | 18 ++++++----
task-sdk/pyproject.toml | 2 +-
uv.lock | 22 +++++++-----
20 files changed, 249 insertions(+), 77 deletions(-)
diff --git a/airflow-core/pyproject.toml b/airflow-core/pyproject.toml
index a843480c11c..ef0755a4243 100644
--- a/airflow-core/pyproject.toml
+++ b/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
diff --git a/airflow-core/src/airflow/api_fastapi/app.py
b/airflow-core/src/airflow/api_fastapi/app.py
index 3c2c94114a5..273a9641fe2 100644
--- a/airflow-core/src/airflow/api_fastapi/app.py
+++ b/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,
diff --git a/airflow-core/src/airflow/api_fastapi/common/http_metrics.py
b/airflow-core/src/airflow/api_fastapi/common/http_metrics.py
index feabd66d4d4..99bfbead1f7 100644
--- a/airflow-core/src/airflow/api_fastapi/common/http_metrics.py
+++ b/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.
diff --git a/airflow-core/src/airflow/api_fastapi/common/telemetry.py
b/airflow-core/src/airflow/api_fastapi/common/telemetry.py
new file mode 100644
index 00000000000..6d7bae34e70
--- /dev/null
+++ b/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,
+}
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/app.py
b/airflow-core/src/airflow/api_fastapi/execution_api/app.py
index bf254e32c4e..21ed44c6764 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/app.py
+++ b/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)
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
index db998c4ad54..f911b4ffab9 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
+++
b/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),
]
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/security.py
b/airflow-core/src/airflow/api_fastapi/execution_api/security.py
index 0d02409d578..ff6daa52c6a 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/security.py
+++ b/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
diff --git a/airflow-core/src/airflow/utils/serve_logs/log_server.py
b/airflow-core/src/airflow/utils/serve_logs/log_server.py
index 92178bf5a24..3726d53a609 100644
--- a/airflow-core/src/airflow/utils/serve_logs/log_server.py
+++ b/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"),
diff --git a/airflow-core/tests/unit/api_fastapi/common/test_http_metrics.py
b/airflow-core/tests/unit/api_fastapi/common/test_http_metrics.py
index af508cfb8b1..4894251df92 100644
--- a/airflow-core/tests/unit/api_fastapi/common/test_http_metrics.py
+++ b/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",
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
index ff1e5e1d038..6ab18c19990 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
+++ b/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(
diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/test_routes.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/test_routes.py
index 7c2e58403ec..89d4ed29bae 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/test_routes.py
+++ b/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)"
diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/test_app.py
b/airflow-core/tests/unit/api_fastapi/execution_api/test_app.py
index 11c056410e4..5f708651294 100644
--- a/airflow-core/tests/unit/api_fastapi/execution_api/test_app.py
+++ b/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")
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/test_token_scope_boundaries.py
b/airflow-core/tests/unit/api_fastapi/execution_api/test_token_scope_boundaries.py
index 48b19c23695..9535ec9522c 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/test_token_scope_boundaries.py
+++
b/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)
diff --git a/airflow-core/tests/unit/api_fastapi/logging/test_decorators.py
b/airflow-core/tests/unit/api_fastapi/logging/test_decorators.py
index 66fb2fe3773..d3a25732410 100644
--- a/airflow-core/tests/unit/api_fastapi/logging/test_decorators.py
+++ b/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)
diff --git a/airflow-core/tests/unit/api_fastapi/test_app.py
b/airflow-core/tests/unit/api_fastapi/test_app.py
index 149179ebd62..5ffcd67354c 100644
--- a/airflow-core/tests/unit/api_fastapi/test_app.py
+++ b/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
[email protected](
+ "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] = [
diff --git a/devel-common/src/tests_common/test_utils/api_fastapi.py
b/devel-common/src/tests_common/test_utils/api_fastapi.py
index 0e19c148614..06ef6a2381b 100644
--- a/devel-common/src/tests_common/test_utils/api_fastapi.py
+++ b/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
diff --git a/providers/edge3/tests/unit/edge3/worker_api/test_app.py
b/providers/edge3/tests/unit/edge3/worker_api/test_app.py
index ec1c54522c9..b06567e6c40 100644
--- a/providers/edge3/tests/unit/edge3/worker_api/test_app.py
+++ b/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:
diff --git a/scripts/in_container/in_container_utils.py
b/scripts/in_container/in_container_utils.py
index 1368f0817c0..5ee8e786c04 100644
--- a/scripts/in_container/in_container_utils.py
+++ b/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"] = {
diff --git a/task-sdk/pyproject.toml b/task-sdk/pyproject.toml
index 7180fc1ccab..aa2876e5e0b 100644
--- a/task-sdk/pyproject.toml
+++ b/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",
diff --git a/uv.lock b/uv.lock
index 7cb99ce80d6..1e9303eb957 100644
--- a/uv.lock
+++ b/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" },
