mirror of https://github.com/tiangolo/fastapi.git
283 lines
8.3 KiB
Python
283 lines
8.3 KiB
Python
"""Tests for lifespan-scoped dependencies (Depends(..., scope="lifespan"))."""
|
|
|
|
from contextlib import asynccontextmanager
|
|
from typing import Annotated
|
|
|
|
import pytest
|
|
from fastapi import 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_lifespan_dependency_decorator_level_dependencies_runs_at_startup() -> None:
|
|
"""Decorator-level dependencies=[Depends(..., scope='lifespan')] run at startup once."""
|
|
started: list[str] = []
|
|
stopped: list[str] = []
|
|
|
|
def lifespan_dep() -> str:
|
|
started.append("lifespan_dep")
|
|
yield "ok"
|
|
stopped.append("lifespan_dep")
|
|
|
|
app = FastAPI()
|
|
|
|
@app.get("/", dependencies=[Depends(lifespan_dep, scope="lifespan")])
|
|
def root() -> dict[str, str]:
|
|
return {"ok": "yes"}
|
|
|
|
with TestClient(app) as client:
|
|
assert started == ["lifespan_dep"]
|
|
r1 = client.get("/")
|
|
r2 = client.get("/")
|
|
assert r1.status_code == 200 and r2.status_code == 200
|
|
assert r1.json() == {"ok": "yes"}
|
|
assert r2.json() == {"ok": "yes"}
|
|
assert started == ["lifespan_dep"]
|
|
|
|
assert stopped == ["lifespan_dep"]
|
|
|
|
|
|
def test_lifespan_dependency_synthetic_request_receive_send() -> None:
|
|
"""Lifespan dep that uses Request.receive covers noop_receive during startup."""
|
|
|
|
async def lifespan_dep(request: Request) -> str:
|
|
await request.receive()
|
|
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_sync_callable() -> None:
|
|
"""Sync (non-gen, non-coroutine) lifespan dep runs via run_in_threadpool (utils 702)."""
|
|
|
|
def sync_lifespan_dep() -> str:
|
|
return "sync_val"
|
|
|
|
app = FastAPI()
|
|
|
|
@app.get("/")
|
|
def root(
|
|
v: Annotated[str, Depends(sync_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": "sync_val"}
|
|
|
|
|
|
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_shared_cache_hit() -> None:
|
|
"""Two lifespan deps B and C both depend on A; second resolution hits cache (utils 687)."""
|
|
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"
|
|
|
|
def lifespan_c(
|
|
a: Annotated[str, Depends(lifespan_a, scope="lifespan")],
|
|
) -> str:
|
|
order.append("c")
|
|
yield a + "-c"
|
|
|
|
app = FastAPI()
|
|
|
|
@app.get("/")
|
|
def root(
|
|
b: Annotated[str, Depends(lifespan_b, scope="lifespan")],
|
|
c: Annotated[str, Depends(lifespan_c, scope="lifespan")],
|
|
) -> dict[str, str]:
|
|
return {"b": b, "c": c}
|
|
|
|
with TestClient(app) as client:
|
|
r = client.get("/")
|
|
assert r.status_code == 200
|
|
assert r.json() == {"b": "a-b", "c": "a-c"}
|
|
assert order == ["a", "b", "c"]
|
|
|
|
|
|
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 # pragma: no cover - never run; raises at app.get("/")(root)
|
|
|
|
def lifespan_dep(
|
|
x: Annotated[int, Depends(request_scoped, scope="request")],
|
|
) -> int:
|
|
return x # pragma: no cover - never run; raises at app.get("/")(root)
|
|
|
|
def root(
|
|
y: Annotated[int, Depends(lifespan_dep, scope="lifespan")],
|
|
) -> dict[str, int]:
|
|
return {"y": y} # pragma: no cover - never run; raises at app.get("/")(root)
|
|
|
|
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" # pragma: no cover - never run; request raises before dep runs
|
|
|
|
sub_app = FastAPI()
|
|
|
|
@sub_app.get("/sub")
|
|
def sub_root(
|
|
x: Annotated[str, Depends(lifespan_dep, scope="lifespan")],
|
|
) -> dict[str, str]:
|
|
return {"x": x} # pragma: no cover - never run; request raises before handler
|
|
|
|
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()
|