forked from fastapi/fastapi
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_dependency_contextvars.py
More file actions
52 lines (37 loc) · 1.53 KB
/
Copy pathtest_dependency_contextvars.py
File metadata and controls
52 lines (37 loc) · 1.53 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
from collections.abc import Awaitable, Callable
from contextvars import ContextVar
from typing import Any
from fastapi import Depends, FastAPI, Request, Response
from fastapi.testclient import TestClient
legacy_request_state_context_var: ContextVar[dict[str, Any] | None] = ContextVar(
"legacy_request_state_context_var", default=None
)
app = FastAPI()
async def set_up_request_state_dependency():
request_state = {"user": "deadpond"}
contextvar_token = legacy_request_state_context_var.set(request_state)
yield request_state
legacy_request_state_context_var.reset(contextvar_token)
@app.middleware("http")
async def custom_middleware(
request: Request, call_next: Callable[[Request], Awaitable[Response]]
):
response = await call_next(request)
response.headers["custom"] = "foo"
return response
@app.get("/user", dependencies=[Depends(set_up_request_state_dependency)])
def get_user():
request_state = legacy_request_state_context_var.get()
assert request_state
return request_state["user"]
client = TestClient(app)
def test_dependency_contextvars():
"""
Check that custom middlewares don't affect the contextvar context for dependencies.
The code before yield and the code after yield should be run in the same contextvar
context, so that request_state_context_var.reset(contextvar_token).
If they are run in a different context, that raises an error.
"""
response = client.get("/user")
assert response.json() == "deadpond"
assert response.headers["custom"] == "foo"