From 9fb0fc81206fd3601ee65caa171bde1e75d3417e Mon Sep 17 00:00:00 2001 From: essentiaMarco <131397104+essentiaMarco@users.noreply.github.com> Date: Fri, 13 Mar 2026 23:39:27 -0700 Subject: [PATCH] refactor(lifespan): align with reviewer feedback, add coverage tests - Fix _collect_lifespan_dependants: handle route-level flat.computed_scope and iterate flat.dependencies (was incorrectly iterating flat) - No pragmas on noop_receive/noop_send; covered by test that uses Request.receive/send in a lifespan dependency - Add test_collect_lifespan_dependants_route_level_scope for route-level lifespan branch - Add test_lifespan_dependency_synthetic_request_receive_send for noop_receive/noop_send coverage - Add test_lifespan_dependency_nested for dependency_cache hit (utils) - Add test_lifespan_dependency_cannot_depend_on_request_scope and test_lifespan_dependency_not_initialized_raises for edge cases Made-with: Cursor --- fastapi/routing.py | 70 ++++++++ tests/test_dependency_lifespan_scope.py | 216 ++++++++++++++++++++++++ 2 files changed, 286 insertions(+) create mode 100644 tests/test_dependency_lifespan_scope.py diff --git a/fastapi/routing.py b/fastapi/routing.py index e2c83aa7b3..f4c943e641 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -202,6 +202,66 @@ def _wrap_gen_lifespan_context( return wrapper +def _collect_lifespan_dependants(router: "APIRouter") -> list[Dependant]: + """Collect all unique lifespan-scoped dependants from router and nested routers.""" + seen: dict[tuple[Any, ...], Dependant] = {} + for route in router.routes: + if isinstance(route, APIRoute): + flat = get_flat_dependant(route.dependant) + if flat.computed_scope == "lifespan": + key = flat.cache_key + if key not in seen: + seen[key] = flat + for d in flat.dependencies: + if d.computed_scope == "lifespan": + key = d.cache_key + if key not in seen: + seen[key] = d + return list(seen.values()) + + +async def _run_lifespan_dependencies( + router: "APIRouter", + dependency_cache: dict[tuple[Any, ...], Any], + lifespan_stack: AsyncExitStack, +) -> None: + """Solve all lifespan-scoped dependencies and fill dependency_cache.""" + from starlette.requests import Request + from starlette.types import Receive, Send + + lifespan_deps = _collect_lifespan_dependants(router) + if not lifespan_deps: + return + synthetic = Dependant(call=None, path="/", dependencies=lifespan_deps) + # Minimal scope so solve_dependencies can run; lifespan_stack used for cleanup. + scope: dict[str, Any] = { + "type": "http", + "path": "/", + "path_params": {}, + "query_string": b"", + "headers": [], + "fastapi_inner_astack": lifespan_stack, + "fastapi_function_astack": lifespan_stack, + } + + async def noop_receive() -> Any: + return {"type": "http.disconnect"} + + async def noop_send(message: Any) -> None: + pass + + request = Request(scope, noop_receive, noop_send) + await solve_dependencies( + request=request, + dependant=synthetic, + body=None, + dependency_cache=dependency_cache, + async_exit_stack=lifespan_stack, + embed_body_fields=False, + solving_lifespan_deps=True, + ) + + def _merge_lifespan_context( original_context: Lifespan[Any], nested_context: Lifespan[Any] ) -> Lifespan[Any]: @@ -450,11 +510,16 @@ def get_request_handler( assert isinstance(async_exit_stack, AsyncExitStack), ( "fastapi_inner_astack not found in request scope" ) + lifespan_cache = getattr( + request.app.state, "fastapi_lifespan_dependency_cache", None + ) + dependency_cache = dict(lifespan_cache) if lifespan_cache else None solved_result = await solve_dependencies( request=request, dependant=dependant, body=body, dependency_overrides_provider=dependency_overrides_provider, + dependency_cache=dependency_cache, async_exit_stack=async_exit_stack, embed_body_fields=embed_body_fields, ) @@ -744,10 +809,15 @@ def get_websocket_app( assert isinstance(async_exit_stack, AsyncExitStack), ( "fastapi_inner_astack not found in request scope" ) + lifespan_cache = getattr( + websocket.app.state, "fastapi_lifespan_dependency_cache", None + ) + dependency_cache = dict(lifespan_cache) if lifespan_cache else None solved_result = await solve_dependencies( request=websocket, dependant=dependant, dependency_overrides_provider=dependency_overrides_provider, + dependency_cache=dependency_cache, async_exit_stack=async_exit_stack, embed_body_fields=embed_body_fields, ) diff --git a/tests/test_dependency_lifespan_scope.py b/tests/test_dependency_lifespan_scope.py new file mode 100644 index 0000000000..dac9fbbc0a --- /dev/null +++ b/tests/test_dependency_lifespan_scope.py @@ -0,0 +1,216 @@ +"""Tests for lifespan-scoped dependencies (Depends(..., scope="lifespan")).""" + +from contextlib import asynccontextmanager +from typing import Annotated + +import pytest +from fastapi import APIRouter, Depends, FastAPI +from fastapi.exceptions import DependencyScopeError +from fastapi.testclient import TestClient +from starlette.requests import Request + + +def test_lifespan_dependency_single_request() -> None: + """Lifespan-scoped dependency is created once and reused across requests.""" + started: list[str] = [] + stopped: list[str] = [] + + def get_db() -> str: + started.append("db") + yield "db_conn" + stopped.append("db") + + app = FastAPI() + + @app.get("/") + def root(db: Annotated[str, Depends(get_db, scope="lifespan")]) -> dict[str, str]: + return {"db": db} + + assert len(started) == 0 + assert len(stopped) == 0 + + with TestClient(app) as client: + assert len(started) == 1, "lifespan dep should start once at app startup" + r1 = client.get("/") + assert r1.status_code == 200 + assert r1.json() == {"db": "db_conn"} + r2 = client.get("/") + assert r2.status_code == 200 + assert r2.json() == {"db": "db_conn"} + assert len(started) == 1, "lifespan dep should not restart per request" + + assert len(stopped) == 1, "lifespan dep should stop once at app shutdown" + + +def test_lifespan_dependency_with_custom_lifespan() -> None: + """Lifespan-scoped dependency runs inside app lifespan and is cleaned up on shutdown.""" + started: list[str] = [] + stopped: list[str] = [] + + @asynccontextmanager + async def lifespan(app: FastAPI): + started.append("lifespan") + yield + stopped.append("lifespan") + + def get_pool() -> str: + started.append("pool") + yield "pool_conn" + stopped.append("pool") + + app = FastAPI(lifespan=lifespan) + + @app.get("/") + def root( + pool: Annotated[str, Depends(get_pool, scope="lifespan")] + ) -> dict[str, str]: + return {"pool": pool} + + with TestClient(app) as client: + assert "lifespan" in started + assert "pool" in started + r = client.get("/") + assert r.status_code == 200 + assert r.json() == {"pool": "pool_conn"} + + assert "pool" in stopped + assert "lifespan" in stopped + + +def test_lifespan_dependency_same_instance_across_requests() -> None: + """The same instance is injected for every request when scope is lifespan.""" + instances: list[object] = [] + + def get_singleton() -> object: + inst = object() + instances.append(inst) + yield inst + + app = FastAPI() + + @app.get("/") + def root( + s: Annotated[object, Depends(get_singleton, scope="lifespan")] + ) -> dict[str, bool]: + return {"is_singleton": len(instances) == 1 and s is instances[0]} + + with TestClient(app) as client: + r1 = client.get("/") + r2 = client.get("/") + assert r1.status_code == 200 and r2.status_code == 200 + assert r1.json()["is_singleton"] is True + assert r2.json()["is_singleton"] is True + assert len(instances) == 1 + + +def test_collect_lifespan_dependants_route_level_scope() -> None: + """Covers _collect_lifespan_dependants when route's flat dependant has computed_scope lifespan.""" + from fastapi.routing import _collect_lifespan_dependants + + router = APIRouter() + + @router.get("/") + def root() -> dict[str, str]: + return {"ok": "yes"} + + route = next(r for r in router.routes if hasattr(r, "dependant")) + # Simulate route-level lifespan scope so the flat.computed_scope == "lifespan" branch is hit + route.dependant.scope = "lifespan" + result = _collect_lifespan_dependants(router) + assert len(result) == 1 + assert result[0].computed_scope == "lifespan" + + +def test_lifespan_dependency_synthetic_request_receive_send() -> None: + """Lifespan dep that uses Request.receive/send covers noop_receive and noop_send during startup.""" + async def lifespan_dep(request: Request) -> str: + await request.receive() + await request.send({"type": "http.response.body"}) + return "ok" + + app = FastAPI() + + @app.get("/") + def root( + v: Annotated[str, Depends(lifespan_dep, scope="lifespan")], + ) -> dict[str, str]: + return {"v": v} + + with TestClient(app) as client: + r = client.get("/") + assert r.status_code == 200 + assert r.json() == {"v": "ok"} + + +def test_lifespan_dependency_nested() -> None: + """Lifespan dep B depending on A covers dependency_cache hit path (utils.py line 685).""" + order: list[str] = [] + + def lifespan_a() -> str: + order.append("a") + yield "a" + + def lifespan_b( + a: Annotated[str, Depends(lifespan_a, scope="lifespan")], + ) -> str: + order.append("b") + yield a + "-b" + + app = FastAPI() + + @app.get("/") + def root( + b: Annotated[str, Depends(lifespan_b, scope="lifespan")], + ) -> dict[str, str]: + return {"b": b} + + with TestClient(app) as client: + r = client.get("/") + assert r.status_code == 200 + assert r.json() == {"b": "a-b"} + assert order == ["a", "b"] + + +def test_lifespan_dependency_cannot_depend_on_request_scope() -> None: + """Lifespan-scoped dependency that depends on request-scoped dep raises.""" + + def request_scoped() -> int: + return 1 + + def lifespan_dep( + x: Annotated[int, Depends(request_scoped, scope="request")], + ) -> int: + return x + + def root( + y: Annotated[int, Depends(lifespan_dep, scope="lifespan")], + ) -> dict[str, int]: + return {"y": y} + + app = FastAPI() + with pytest.raises(DependencyScopeError) as exc_info: + app.get("/")(root) + assert "lifespan" in str(exc_info.value) and "cannot depend" in str(exc_info.value) + + +def test_lifespan_dependency_not_initialized_raises() -> None: + """Request that needs a lifespan dep which was not run (e.g. mounted sub-app) raises.""" + + def lifespan_dep() -> str: + yield "conn" + + sub_app = FastAPI() + + @sub_app.get("/sub") + def sub_root( + x: Annotated[str, Depends(lifespan_dep, scope="lifespan")], + ) -> dict[str, str]: + return {"x": x} + + main_app = FastAPI() + main_app.mount("/mounted", sub_app) + + with TestClient(main_app) as client: + with pytest.raises(DependencyScopeError) as exc_info: + client.get("/mounted/sub") + assert "lifespan" in str(exc_info.value).lower()