This is an automated email from the ASF dual-hosted git repository. vatsrahul1001 pushed a commit to branch backport-b70a439-v3-4-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 999fa53c4577d75d9693f3c7d655b105fd312e07 Author: Rahul Vats <[email protected]> AuthorDate: Mon Oct 26 22:36:04 2026 -0600 [v3-4-test] Allow FastAPI 0.137+ by requiring cadwyn 8.1 (#75689) 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 480707fd6cc..08d22de3f96 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" },
