Coverage for tests / test_dependency_contextvars.py: 100%

1from collections.abc import Awaitable 4 ctx1abcd

2from contextvars import ContextVar 4 ctx1abcd

3from typing import Any, Callable, Optional 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[Optional[dict[str, Any]]] = 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"} 3 ctx1efg

17 contextvar_token = legacy_request_state_context_var.set(request_state) 3 ctx1efg

18 yield request_state 3 ctx1efg

19 legacy_request_state_context_var.reset(contextvar_token) 3 ctx1efg

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) 3 ctx1efg

27 response.headers["custom"] = "foo" 3 ctx1efg

28 return response 3 ctx1efg

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() 3 ctx1efg

34 assert request_state 3 ctx1efg

35 return request_state["user"] 3 ctx1efg

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") 3 ctx1efg

51 assert response.json() == "deadpond" 3 ctx1efg

52 assert response.headers["custom"] == "foo" 3 ctx1efg