将在端点中可用的 FastAPI 保存上下文

Question

我的目标是从请求中提取一些信息并将其放在某个全局上下文中。 我尝试的第一件事是执行此操作的中间件函数

user_ctx: ContextVar[typing.Optional[UUID]] = ContextVar('user_ctx', default=None)

def save_user_ctx(request: Request):
    uid = ... # logic that extracts user uid from request header
    user_ctx.set(uid)

我将它添加到另一个中间件函数中，例如my_middleware。

所以它确实设置了上下文，但据我所知中间件在另一个线程中工作，所以如果我尝试类似

@router.post('/some_url', response_model=SomeResponseModel)
def some_url(data: SomeRequestModel, auth_checker = Depends(my_middleware)):
    user_ctx.get() # >> None

auth_checker 拥有所有身份验证信息，因此中间件可以正常工作。但是uid = user_ctx.get() 返回None。

这只是一个例子。请

我想要实现的是将请求中的一些上下文存储在自动为端点提供服务的线程中。我的意思是不写ctx.set(smth) 作为每个端点的第一行。

我也试过写装饰器

def save_request_ctx(func):
    @wraps(func)
    def wrap(*args, **kwargs):
        ctx.set(...)
        return func(*args, **kwargs)
    return wrap

但如果端点没有request: Request 作为参数，它不会出现在args 或kwargs 中。因此，此外，如果您可以告诉如何强制每个端点将 request 作为参数而不在每个端点函数中手动添加它 - 它也将被接受为解决方案。

如果我对我想要的解释很复杂，我会尝试澄清它。

Answer 1

您可能想使用我的starlette-context。

为了达到您想要的效果，最好将ContextMiddleware 子类化并覆盖set_context 方法。然后返回的内容将在 context 对象中可用，您可以在任何地方导入该对象。

示例代码

from starlette_context.middleware import ContextMiddleware


class YourMiddleware(ContextMiddleware):
    async def set_context(self, request: Request) -> dict:
        return {"from_middleware": True}

注册此中间件后，您可以在视图/记录器中导入

from starlette_context import context

并使用context["from_middleware"]。

请注意，我为最常见的用例准备了“插件”。如果您想在应用中的任何位置访问请求/关联 ID，只需使用

初始化中间件

from starlette_context import context, plugins
from starlette_context.middleware import ContextMiddleware

middleware = [
    Middleware(
        ContextMiddleware,
        plugins=(plugins.RequestIdPlugin(), plugins.CorrelationIdPlugin()),
    )
]

app = Starlette(debug=True, middleware=middleware)

FastAPI 使用 Starlette 接口，所以你应该很高兴。

关于

但是如果端点没有 request: Request 作为它得到的参数 - 它不会出现在 args 或 kwargs 中。因此，此外，如果您可以告诉如何强制每个端点将请求作为参数而不在每个端点函数中手动添加它 - 它也将被接受为解决方案。

我无法回答这个问题，因为我从未尝试过使用没有请求参数的视图。

编辑：我添加了不需要请求对象的RawContextMiddleware。