From f434abd5e3371aac0997119bd70d87829256c680 Mon Sep 17 00:00:00 2001 From: Revanth Ch Date: Mon, 22 Jun 2026 02:20:57 -0500 Subject: [PATCH] Restore FastAPI 0.137 support in the Task Execution API --- airflow-core/pyproject.toml | 8 +- .../airflow/api_fastapi/execution_api/app.py | 33 +++----- .../execution_api/routes/__init__.py | 82 ++++++++++++------- .../api_fastapi/execution_api/test_app.py | 14 ---- uv.lock | 16 ++-- 5 files changed, 74 insertions(+), 79 deletions(-) diff --git a/airflow-core/pyproject.toml b/airflow-core/pyproject.toml index 31a88734c88fd..0d5aac0805fa2 100644 --- a/airflow-core/pyproject.toml +++ b/airflow-core/pyproject.toml @@ -84,7 +84,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>=7.1.0", "colorlog>=6.8.2", "cron-descriptor>=1.2.24", "croniter>=2.0.2", @@ -95,11 +95,7 @@ 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", + "fastapi[standard-no-fastapi-cloud-cli]>=0.137.1", "uvicorn>=0.37.0", "starlette>=1.0.1", "httpx>=0.25.0", 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 88c7d23fff26a..219350a6a4540 100644 --- a/airflow-core/src/airflow/api_fastapi/execution_api/app.py +++ b/airflow-core/src/airflow/api_fastapi/execution_api/app.py @@ -259,30 +259,9 @@ async def _extract_w3c_trace_context( otel_context.detach(token) -def _inject_trace_context_dep(routes, mode: str) -> None: - dep = Depends(_extract_w3c_trace_context) - for route in routes: - 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) - - def create_task_execution_api_app() -> FastAPI: """Create FastAPI app for task execution API.""" - from airflow.api_fastapi.execution_api.routes import execution_api_router + from airflow.api_fastapi.execution_api.routes import build_execution_api_router from airflow.api_fastapi.execution_api.versions import bundle from airflow.configuration import conf @@ -305,8 +284,16 @@ def custom_generate_unique_id(route: APIRoute): app.add_middleware(CorrelationIdMiddleware) app.add_middleware(JWTReissueMiddleware) + # FastAPI (>=0.137) freezes a router's dependencies into each route at include time, so the + # trace-context dependency has to be supplied while the router tree is assembled. "unsafe-always" + # extracts on every request (before auth); "only-authenticated" extracts only after a successful + # require_auth; "never" extracts nothing. mode = conf.get("execution_api", "otel_trace_propagation", fallback="only-authenticated") - _inject_trace_context_dep(execution_api_router.routes, mode) + trace_context_dep = Depends(_extract_w3c_trace_context) + execution_api_router = build_execution_api_router( + pre_auth_dependencies=[trace_context_dep] if mode == "unsafe-always" else (), + post_auth_dependencies=[trace_context_dep] if mode == "only-authenticated" else (), + ) app.generate_and_include_versioned_routers(execution_api_router) diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py b/airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py index 7b19f3ddd3055..f7e9fd035d4e6 100644 --- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py +++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py @@ -16,6 +16,9 @@ # under the License. from __future__ import annotations +from collections.abc import Sequence +from typing import Any + from cadwyn import VersionedAPIRouter from fastapi import APIRouter, Security @@ -37,34 +40,57 @@ ) from airflow.api_fastapi.execution_api.security import require_auth -execution_api_router = APIRouter() -# health.router declares its full paths ("/health", "/health/ping") and is included without a -# prefix, unlike the routers below. A root route registered as @router.get("") under an include-time -# prefix=... raises "Prefix and path cannot be both empty" once FastAPI switched to lazy router -# inclusion (>=0.137); see https://github.com/apache/airflow/issues/68562. Don't reintroduce a prefix here. -execution_api_router.include_router(health.router, tags=["Health"]) -# _Every_ single endpoint under here must be authenticated. Some do further checks on top of these -authenticated_router = VersionedAPIRouter(dependencies=[Security(require_auth)]) # type: ignore[list-item] +def build_execution_api_router( + *, + pre_auth_dependencies: Sequence[Any] = (), + post_auth_dependencies: Sequence[Any] = (), +) -> APIRouter: + """ + Assemble the Task Execution API router tree. -authenticated_router.include_router(assets.router, prefix="/assets", tags=["Assets"]) -authenticated_router.include_router(asset_events.router, prefix="/asset-events", tags=["Asset Events"]) -authenticated_router.include_router( - connection_tests.router, prefix="/connection-tests", tags=["Connection Tests"] -) -authenticated_router.include_router(connections.router, prefix="/connections", tags=["Connections"]) -authenticated_router.include_router(dag_runs.router, prefix="/dag-runs", tags=["Dag Runs"]) -authenticated_router.include_router(dags.router, prefix="/dags", tags=["Dags"]) -authenticated_router.include_router(task_instances.router, prefix="/task-instances", tags=["Task Instances"]) -authenticated_router.include_router( - task_reschedules.router, prefix="/task-reschedules", tags=["Task Reschedules"] -) -authenticated_router.include_router(variables.router, prefix="/variables", tags=["Variables"]) -authenticated_router.include_router(xcoms.router, prefix="/xcoms", tags=["XComs"]) -authenticated_router.include_router(hitl.router, prefix="/hitlDetails", tags=["Human in the Loop"]) -authenticated_router.include_router(task_state_store.router, prefix="/store/ti", tags=["Task State Store"]) -authenticated_router.include_router( - asset_state_store.router, prefix="/store/asset", tags=["Asset State Store"] -) + ``pre_auth_dependencies`` run before authentication on every route (health included); + ``post_auth_dependencies`` run after ``require_auth`` and therefore only for successfully + authenticated requests. They must be supplied here, at build time: FastAPI (>=0.137) snapshots a + router's dependencies into each route when ``include_router`` is called, so dependencies attached + afterwards to a shared, already-assembled router never take effect. + """ + # _Every_ single endpoint under here must be authenticated. Some do further checks on top of these. + authenticated_router = VersionedAPIRouter( + dependencies=[Security(require_auth), *post_auth_dependencies] # type: ignore[list-item] + ) + + authenticated_router.include_router(assets.router, prefix="/assets", tags=["Assets"]) + authenticated_router.include_router(asset_events.router, prefix="/asset-events", tags=["Asset Events"]) + authenticated_router.include_router( + connection_tests.router, prefix="/connection-tests", tags=["Connection Tests"] + ) + authenticated_router.include_router(connections.router, prefix="/connections", tags=["Connections"]) + authenticated_router.include_router(dag_runs.router, prefix="/dag-runs", tags=["Dag Runs"]) + authenticated_router.include_router(dags.router, prefix="/dags", tags=["Dags"]) + authenticated_router.include_router( + task_instances.router, prefix="/task-instances", tags=["Task Instances"] + ) + authenticated_router.include_router( + task_reschedules.router, prefix="/task-reschedules", tags=["Task Reschedules"] + ) + authenticated_router.include_router(variables.router, prefix="/variables", tags=["Variables"]) + authenticated_router.include_router(xcoms.router, prefix="/xcoms", tags=["XComs"]) + authenticated_router.include_router(hitl.router, prefix="/hitlDetails", tags=["Human in the Loop"]) + authenticated_router.include_router( + task_state_store.router, prefix="/store/ti", tags=["Task State Store"] + ) + authenticated_router.include_router( + asset_state_store.router, prefix="/store/asset", tags=["Asset State Store"] + ) -execution_api_router.include_router(authenticated_router) + execution_api_router = APIRouter() + # health.router declares its full paths ("/health", "/health/ping") and is included without a + # prefix, unlike the routers above. A root route registered as @router.get("") under an include-time + # prefix=... raises "Prefix and path cannot be both empty" once FastAPI switched to lazy router + # inclusion (>=0.137); see https://github.com/apache/airflow/issues/68562. Don't reintroduce a prefix here. + execution_api_router.include_router( + health.router, tags=["Health"], dependencies=list(pre_auth_dependencies) + ) + execution_api_router.include_router(authenticated_router, dependencies=list(pre_auth_dependencies)) + return execution_api_router 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 a5f371e565fa4..bce3d3a270e28 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 @@ -190,20 +190,6 @@ def test_multiple_requests_with_different_correlation_ids(self, client): class TestTraceContextPropagation: """Exercise ``execution_api.otel_trace_propagation`` on the real Execution API app.""" - @pytest.fixture(autouse=True) - def _restore_router_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) - } - yield - for route in execution_api_router.routes: - if isinstance(route, APIRoute): - route.dependencies[:] = snapshot[id(route)] - @staticmethod def _build_app(mode: str): with conf_vars({("execution_api", "otel_trace_propagation"): mode}): diff --git a/uv.lock b/uv.lock index 8d5f31fa71947..720adb14d9161 100644 --- a/uv.lock +++ b/uv.lock @@ -2059,7 +2059,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 = ">=7.1.0" }, { name = "colorlog", specifier = ">=6.8.2" }, { name = "cron-descriptor", specifier = ">=1.2.24" }, { name = "croniter", specifier = ">=2.0.2" }, @@ -2067,7 +2067,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.137.1" }, { 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" }, @@ -10211,7 +10211,7 @@ wheels = [ [[package]] name = "cadwyn" -version = "7.0.0" +version = "7.1.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "backports-strenum", marker = "python_full_version < '3.11'" }, @@ -10222,9 +10222,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/eb/74/b557960408e3577d57eb9b5a677d081bc36d2fc5b02cfd15791de35a3257/cadwyn-7.1.0.tar.gz", hash = "sha256:0272361a7c97723a9639258a7e1f02c7832b24b7929ac3f3be805eb9de53a550", size = 666212, upload-time = "2026-06-15T18:42:00.757Z" } 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/b8/53/a7d4153847692d59de6cfd7b41a26c66dd633b2bda5e098305819f50c0ad/cadwyn-7.1.0-py3-none-any.whl", hash = "sha256:a72859c326e10dce63cbe360bf0aa88ffd2785be7024b3baafe49be25e2be3d6", size = 62227, upload-time = "2026-06-15T18:41:59.496Z" }, ] [[package]] @@ -11675,7 +11675,7 @@ wheels = [ [[package]] name = "fastapi" -version = "0.136.3" +version = "0.137.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "annotated-doc" }, @@ -11684,9 +11684,9 @@ dependencies = [ { 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/d5/b1/e5b92c59d2c37817e77c1a8c2fc1f79cdcc04c68253e5406b43e3204cba7/fastapi-0.137.1.tar.gz", hash = "sha256:822360704230d9533d8d9475399613525968aa2f0b5bd2a3ccc9f18c88fd541c", size = 408293, upload-time = "2026-06-15T11:28:20.79Z" } 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/da/35/380b9a5922f4340e51c309cde09e5bd32e62f02302971bee30dc15aa0624/fastapi-0.137.1-py3-none-any.whl", hash = "sha256:64f6983c59e45c4b9fdc44e57cb8035c2451ee91ea8e8ec042aca37de7cf6b69", size = 121877, upload-time = "2026-06-15T11:28:19.523Z" }, ] [package.optional-dependencies]