From ee39a581ef5049e202c6dba660b6af0efa761608 Mon Sep 17 00:00:00 2001 From: shaealh Date: Mon, 13 Jul 2026 03:05:56 -0700 Subject: [PATCH 1/3] Migrate to FastAPI lazy router inclusion --- airflow-core/pyproject.toml | 8 +-- .../airflow/api_fastapi/execution_api/app.py | 29 ++------ .../execution_api/routes/__init__.py | 71 +++++++++++-------- .../core_api/routes/test_routes.py | 33 +++++++-- .../api_fastapi/execution_api/test_app.py | 14 ---- .../test_token_scope_boundaries.py | 17 ++--- uv.lock | 16 ++--- 7 files changed, 96 insertions(+), 92 deletions(-) diff --git a/airflow-core/pyproject.toml b/airflow-core/pyproject.toml index 40d18944f3a29..c6d3341f6aabb 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.2", "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 5a87cdb81e04c..bd6e58f3db45c 100644 --- a/airflow-core/src/airflow/api_fastapi/execution_api/app.py +++ b/airflow-core/src/airflow/api_fastapi/execution_api/app.py @@ -264,30 +264,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(lifespan: svcs.fastapi.lifespan = lifespan) -> 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 @@ -311,7 +290,11 @@ def custom_generate_unique_id(route: APIRoute): app.add_middleware(JWTReissueMiddleware) 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..5a7cb05bb5f3c 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,8 +16,11 @@ # under the License. from __future__ import annotations +from collections.abc import Sequence + from cadwyn import VersionedAPIRouter from fastapi import APIRouter, Security +from fastapi.params import Depends as DependsParam from airflow.api_fastapi.execution_api.routes import ( asset_events, @@ -37,34 +40,46 @@ ) 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[DependsParam] = (), + post_auth_dependencies: Sequence[DependsParam] = (), +) -> APIRouter: + """Assemble a fresh Task Execution API router tree with dependencies applied at build time.""" + 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"] -) + 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 routes are intentionally outside authentication. FastAPI 0.137+ snapshots router + # dependencies during lazy inclusion, so both dependency layers must be supplied 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/core_api/routes/test_routes.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/test_routes.py index 95734bdaa37f4..a3dd9a5fd1ef0 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 @@ -16,6 +16,9 @@ # under the License. from __future__ import annotations +from fastapi import APIRouter +from fastapi.routing import iter_route_contexts + from airflow.api_fastapi.core_api.routes.public import authenticated_router, public_router # Set of paths that are allowed to be accessible without authentication @@ -27,21 +30,41 @@ } +def _get_paths_excluded_from_authentication(public: APIRouter, authenticated: APIRouter) -> set[str]: + public_paths = {route.path for route in iter_route_contexts(public.routes) if route.path is not None} + authenticated_paths = { + f"{public.prefix}{route.path}" + for route in iter_route_contexts(authenticated.routes) + if route.path is not None + } + return public_paths - authenticated_paths + + 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_authenticated_router = { - f"{public_router.prefix}{route.path}" for route in authenticated_router.routes + assert _get_paths_excluded_from_authentication(public_router, authenticated_router) == NO_AUTH_PATHS + + +def test_no_auth_routes_guard_detects_unprotected_route(): + """Prove the guard reports a route accidentally registered outside the authenticated router.""" + test_public_router = APIRouter(prefix="/api/v2") + test_authenticated_router = APIRouter() + + @test_public_router.get("/unprotected") + def unprotected_route(): + return None + + assert _get_paths_excluded_from_authentication(test_public_router, test_authenticated_router) == { + "/api/v2/unprotected" } - 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_route_contexts(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 d4b9ce5ac88d7..46b29b827aff1 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 @@ -233,20 +233,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/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 bbf2be8704f46..05eefbff6ae14 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 @@ -37,9 +37,9 @@ from __future__ import annotations import pytest -from fastapi.routing import APIRoute +from fastapi.routing import APIRoute, iter_route_contexts -from airflow.api_fastapi.execution_api.routes import execution_api_router +from airflow.api_fastapi.execution_api.routes import build_execution_api_router # Routes that intentionally deviate from the default (execution-only) policy. # Any route NOT listed here must accept only {"execution"}. @@ -55,12 +55,13 @@ 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 + router = build_execution_api_router() + for route_context in iter_route_contexts(router.routes): + if isinstance(route_context.route, APIRoute): + allowed_tokens = set(getattr(route_context.original_route, "allowed_token_types", {"execution"})) + if route_context.methods: + for method in route_context.methods: + policy_map[f"{method} {route_context.path}"] = allowed_tokens return policy_map diff --git a/uv.lock b/uv.lock index 2973c8da2681a..deece2f29ee41 100644 --- a/uv.lock +++ b/uv.lock @@ -2072,7 +2072,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" }, @@ -2080,7 +2080,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.2" }, { 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" }, @@ -10327,7 +10327,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'" }, @@ -10338,9 +10338,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]] @@ -11776,7 +11776,7 @@ wheels = [ [[package]] name = "fastapi" -version = "0.136.3" +version = "0.139.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "annotated-doc" }, @@ -11785,9 +11785,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/d3/af/a5f50ccfa659ec1802cb4ca842c23f06d906a8cc9aef6016a2caeea3d4ed/fastapi-0.139.0.tar.gz", hash = "sha256:99ab7b2d92223c76d6cf10757ab3f89d45b38267fc20b2a136cf02f6beac3145", size = 423016, upload-time = "2026-07-01T16:35:33.436Z" } 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/9e/7c/8e3c6ad324ea5cb36604fc3f968554887891c316d9dfde57761611d907ad/fastapi-0.139.0-py3-none-any.whl", hash = "sha256:cf15e1e9e667ddb0ad63811e60bd11390d1aac838ca4a7a23f421807b2308189", size = 130339, upload-time = "2026-07-01T16:35:32.19Z" }, ] [package.optional-dependencies] From b6187e20a33156f8fc7495b9bb7584428a08be23 Mon Sep 17 00:00:00 2001 From: shaealh Date: Mon, 13 Jul 2026 04:06:21 -0700 Subject: [PATCH 2/3] Fix OpenAPI generation with lazy routers --- scripts/in_container/in_container_utils.py | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/scripts/in_container/in_container_utils.py b/scripts/in_container/in_container_utils.py index 1368f0817c0bd..a113ac1795dde 100644 --- a/scripts/in_container/in_container_utils.py +++ b/scripts/in_container/in_container_utils.py @@ -67,13 +67,19 @@ 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 APIRoute, iter_route_contexts if only_ui: - for route in app.routes: - if not isinstance(route, APIRoute): - continue - route.include_in_schema = route.path.startswith("/ui/") + routes = [ + route_context + for route_context in iter_route_contexts(app.routes) + if isinstance(route_context.original_route, APIRoute) and route_context.path.startswith("/ui/") + ] + # UI routers are excluded from the public schema, so opt their materialized contexts back in. + for route_context in routes: + route_context._effective_route.include_in_schema = True + else: + routes = app.routes 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"] = { From 04fbf4c7ee411d5458b3e07ab110b963b6fa1532 Mon Sep 17 00:00:00 2001 From: shaealh Date: Tue, 14 Jul 2026 20:46:50 -0700 Subject: [PATCH 3/3] Fix OpenAPI helper typing --- scripts/in_container/in_container_utils.py | 23 +++++++++++++++------- 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/scripts/in_container/in_container_utils.py b/scripts/in_container/in_container_utils.py index a113ac1795dde..ead3becc2fb1e 100644 --- a/scripts/in_container/in_container_utils.py +++ b/scripts/in_container/in_container_utils.py @@ -20,15 +20,18 @@ import subprocess import sys import textwrap +from collections.abc import Sequence from contextlib import contextmanager from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Protocol, cast import rich_click as click 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" @@ -41,6 +44,10 @@ AIRFLOW_DIST_PATH = Path("/dist") +class _OpenAPIRoute(Protocol): + include_in_schema: bool + + @contextmanager def ci_group(group_name: str, github_actions: bool): if github_actions: @@ -69,17 +76,19 @@ def generate_openapi_file(app: FastAPI, file_path: Path, prefix: str = "", only_ from fastapi.openapi.utils import get_openapi from fastapi.routing import APIRoute, iter_route_contexts + routes: Sequence[BaseRoute | RouteContext] = app.routes if only_ui: - routes = [ + route_contexts = [ route_context for route_context in iter_route_contexts(app.routes) - if isinstance(route_context.original_route, APIRoute) and route_context.path.startswith("/ui/") + if isinstance(route_context.original_route, APIRoute) + and (route_path := route_context.path) is not None + and route_path.startswith("/ui/") ] # UI routers are excluded from the public schema, so opt their materialized contexts back in. - for route_context in routes: - route_context._effective_route.include_in_schema = True - else: - routes = app.routes + for route_context in route_contexts: + cast("_OpenAPIRoute", route_context._effective_route).include_in_schema = True + routes = route_contexts with file_path.open("w+") as f: openapi_schema = get_openapi(