This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 0ff47dc6658 [v3-4-test] Allow FastAPI 0.137+ by requiring cadwyn 8.1
(#75689) (#75715)
0ff47dc6658 is described below
commit 0ff47dc665847e0d061c7e9d71a9111f825740bc
Author: Rahul Vats <[email protected]>
AuthorDate: Tue Oct 27 13:17:54 2026 +0530
[v3-4-test] Allow FastAPI 0.137+ by requiring cadwyn 8.1 (#75689) (#75715)
Manual backport: pruned subset of #75689 excluding the #64523 (REST
metrics) and #74370 (dag-processor scoped Execution API) pieces that are not on
v3-4-test. Keeps the fastapi>=0.142/cadwyn>=8.1 bump + native-OTel-disable +
lazy-router adaptations.
(cherry picked from commit b70a439a23827a140ed75498f39c429089f723b5)
---
airflow-core/pyproject.toml | 9 ++---
airflow-core/src/airflow/api_fastapi/app.py | 2 ++
.../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 +-
.../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 +++++++-----
18 files changed, 213 insertions(+), 75 deletions(-)
diff --git a/airflow-core/pyproject.toml b/airflow-core/pyproject.toml
index 7ba28632434..bae2fb15408 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 974fb2cee75..bf8f28b4b5d 100644
--- a/airflow-core/src/airflow/api_fastapi/app.py
+++ b/airflow-core/src/airflow/api_fastapi/app.py
@@ -31,6 +31,7 @@ from airflow import settings
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.telemetry import NATIVE_TELEMETRY_DISABLED
from airflow.api_fastapi.core_api.app import (
init_config,
init_flask_plugins,
@@ -134,6 +135,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/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 0e2271e1282..7f1fe51cc76 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
@@ -260,25 +261,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(
@@ -303,6 +309,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.",
@@ -311,13 +318,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 8a1deb203fd..45bf242fad2 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 2c121b6e4ca..f209fe2da02 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/security.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/security.py
@@ -54,11 +54,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/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 88d210d87f0..7049fc2a60f 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,
_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 195503f2bf0..308de1176d8 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
@@ -41,6 +41,8 @@ from fastapi.routing import APIRoute
from airflow.api_fastapi.execution_api.routes import execution_api_router
+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]] = {
@@ -54,16 +56,25 @@ NON_DEFAULT_TOKEN_POLICY: dict[str, set[str]] = {
}
+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 iter_api_routes(execution_api_router.routes)
+ for method in route.methods or ()
+ }
+
+
def _all_route_policies() -> dict[str, set[str]]:
"""Return a map of all API routes and their allowed token types."""
- policy_map: dict[str, set[str]] = {}
- for route in execution_api_router.routes:
- if isinstance(route, APIRoute):
- allowed_tokens = set(getattr(route, "allowed_token_types",
{"execution"}))
- if route.methods:
- for method in route.methods:
- policy_map[f"{method} {route.path}"] = allowed_tokens
- return policy_map
+ return {
+ key: set(getattr(_get_original_route(route), "allowed_token_types",
{"execution"}))
+ for key, route in _get_api_routes().items()
+ }
class TestTokenScopeBoundaries:
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 59ccb87007c..82e0674bafe 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
@@ -190,6 +193,28 @@ def
test_access_log_middleware_installed_outermost_for_every_apps_selection(apps
assert installed[0] is HttpAccessLogMiddleware
[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 c6cafbe75bc..7b046a636f3 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 5a6ddbc9617..3ef22961f9b 100644
--- a/uv.lock
+++ b/uv.lock
@@ -2142,7 +2142,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" },
@@ -2150,7 +2150,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" },
@@ -9458,7 +9458,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"
},
@@ -10901,7 +10901,7 @@ wheels = [
[[package]]
name = "cadwyn"
-version = "7.0.0"
+version = "8.1.0"
source = { registry = "https://pypi.org/simple"; }
dependencies = [
{ name = "fastapi" },
@@ -10911,9 +10911,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]]
@@ -12577,18 +12577,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]
@@ -12596,7 +12597,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" },
