Coverage for tests/test_dependency_contextvars.py: 100%

1from contextvars import ContextVar 7 ctx1abcdefg

2from typing import Any, Awaitable, Callable, Dict, Optional 7 ctx1abcdefg

3 

4from fastapi import Depends, FastAPI, Request, Response 7 ctx1abcdefg

5from fastapi.testclient import TestClient 7 ctx1abcdefg

6 

7legacy_request_state_context_var: ContextVar[Optional[Dict[str, Any]]] = ContextVar( 7 ctx1abcdefg

8 "legacy_request_state_context_var", default=None 

9) 

10 

11app = FastAPI() 7 ctx1abcdefg

12 

13 

14async def set_up_request_state_dependency(): 7 ctx1abcdefg

15 request_state = {"user": "deadpond"} 7 ctx1hijklmn

16 contextvar_token = legacy_request_state_context_var.set(request_state) 7 ctx1hijklmn

17 yield request_state 7 ctx1hijklmn

18 legacy_request_state_context_var.reset(contextvar_token) 7 ctx1hijklmn

19 

20 

21@app.middleware("http") 7 ctx1abcdefg

22async def custom_middleware( 7 ctx1abcdefg

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

24): 

25 response = await call_next(request) 7 ctx1hijklmn

26 response.headers["custom"] = "foo" 7 ctx1hijklmn

27 return response 7 ctx1hijklmn

28 

29 

30@app.get("/user", dependencies=[Depends(set_up_request_state_dependency)]) 7 ctx1abcdefg

31def get_user(): 7 ctx1abcdefg

32 request_state = legacy_request_state_context_var.get() 7 ctx1hijklmn

33 assert request_state 7 ctx1hijklmn

34 return request_state["user"] 7 ctx1hijklmn

35 

36 

37client = TestClient(app) 7 ctx1abcdefg

38 

39 

40def test_dependency_contextvars(): 7 ctx1abcdefg

41 """ 

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

43 

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

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

46 

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

48 """ 

49 response = client.get("/user") 7 ctx1hijklmn

50 assert response.json() == "deadpond" 7 ctx1hijklmn

51 assert response.headers["custom"] == "foo" 7 ctx1hijklmn