fastapi

1
from contextvars import ContextVar
2
from typing import Any, Awaitable, Callable, Dict, Optional
3

4
from fastapi import Depends, FastAPI, Request, Response
5
from fastapi.testclient import TestClient
6

7
legacy_request_state_context_var: ContextVar[Optional[Dict[str, Any]]] = ContextVar(
8
    "legacy_request_state_context_var", default=None
9
)
10

11
app = FastAPI()
12

13

14
async def set_up_request_state_dependency():
15
    request_state = {"user": "deadpond"}
16
    contextvar_token = legacy_request_state_context_var.set(request_state)
17
    yield request_state
18
    legacy_request_state_context_var.reset(contextvar_token)
19

20

21
@app.middleware("http")
22
async def custom_middleware(
23
    request: Request, call_next: Callable[[Request], Awaitable[Response]]
24
):
25
    response = await call_next(request)
26
    response.headers["custom"] = "foo"
27
    return response
28

29

30
@app.get("/user", dependencies=[Depends(set_up_request_state_dependency)])
31
def get_user():
32
    request_state = legacy_request_state_context_var.get()
33
    assert request_state
34
    return request_state["user"]
35

36

37
client = TestClient(app)
38

39

40
def test_dependency_contextvars():
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")
50
    assert response.json() == "deadpond"
51
    assert response.headers["custom"] == "foo"
52