Coverage for tests / test_dependency_contextvars.py: 100%

1from collections.abc import Awaitable, Callable 4 ctx1abcd

2from contextvars import ContextVar 4 ctx1abcd

3from typing import Any 4 ctx1abcd

4 

5from fastapi import Depends, FastAPI, Request, Response 4 ctx1abcd

6from fastapi.testclient import TestClient 4 ctx1abcd

7 

8legacy_request_state_context_var: ContextVar[dict[str, Any] | None] = ContextVar( 4 ctx1abcd

9 "legacy_request_state_context_var", default=None 

10) 

11 

12app = FastAPI() 4 ctx1abcd

13 

14 

15async def set_up_request_state_dependency(): 4 ctx1abcd

16 request_state = {"user": "deadpond"} 4 ctx1efgh

17 contextvar_token = legacy_request_state_context_var.set(request_state) 4 ctx1efgh

18 yield request_state 4 ctx1efgh

19 legacy_request_state_context_var.reset(contextvar_token) 4 ctx1efgh

20 

21 

22@app.middleware("http") 4 ctx1abcd

23async def custom_middleware( 4 ctx1abcd

24 request: Request, call_next: Callable[[Request], Awaitable[Response]] 

25): 

26 response = await call_next(request) 4 ctx1efgh

27 response.headers["custom"] = "foo" 4 ctx1efgh

28 return response 4 ctx1efgh

29 

30 

31@app.get("/user", dependencies=[Depends(set_up_request_state_dependency)]) 4 ctx1abcd

32def get_user(): 4 ctx1abcd

33 request_state = legacy_request_state_context_var.get() 4 ctx1efgh

34 assert request_state 4 ctx1efgh

35 return request_state["user"] 4 ctx1efgh

36 

37 

38client = TestClient(app) 4 ctx1abcd

39 

40 

41def test_dependency_contextvars(): 4 ctx1abcd

42 """ 

43 Check that custom middlewares don't affect the contextvar context for dependencies. 

44 

45 The code before yield and the code after yield should be run in the same contextvar 

46 context, so that request_state_context_var.reset(contextvar_token). 

47 

48 If they are run in a different context, that raises an error. 

49 """ 

50 response = client.get("/user") 4 ctx1efgh

51 assert response.json() == "deadpond" 4 ctx1efgh

52 assert response.headers["custom"] == "foo" 4 ctx1efgh