FastAPI与Beanie实现scoped当前用户注入及数据库事件赋值问题

问题原因 #

你尝试在Beanie的@before_event方法中直接使用FastAPI的Depends注入当前用户，但Beanie的事件触发机制不在FastAPI的请求依赖注入上下文范围内，导致current_user无法被正确注入，最终传入last_modified_by的是None，触发Pydantic的类型验证错误。

解决方案 #

使用Python标准库的contextvars来存储当前请求的用户信息，它能在异步环境中安全隔离不同请求的上下文数据，让Beanie的事件方法可以获取到当前请求的用户。

具体实现步骤 #

1. 定义上下文变量

from contextvars import ContextVar

# 定义上下文变量存储当前用户，默认值为None
current_user_var: ContextVar[str | None] = ContextVar("current_user", default=None)

2. 修改用户依赖，设置上下文变量

from fastapi import Header, Depends, HTTPException

async def get_current_user(x_user: str | None = Header(None)) -> str:
    if not x_user:
        raise HTTPException(status_code=401, detail="未提供用户标识")
    # 将当前用户存入上下文变量
    current_user_var.set(x_user)
    return x_user

3. 修改Beanie实体的事件方法

from beanie import Document, Replace, before_event

class DataEntity(Document):
    name: str
    last_modified_by: str

    @before_event(Replace)
    def set_version_data(self):
        # 从上下文变量中获取当前用户
        current_user = current_user_var.get()
        if current_user:
            self.last_modified_by = current_user
        else:
            # 根据业务需求处理无用户的情况，比如抛出错误或设置默认值
            raise ValueError("无法获取当前用户，无法更新last_modified_by字段")

额外说明 #

• 确保所有需要自动设置last_modified_by的接口都依赖get_current_user，这样上下文变量才会被正确赋值。
• 如果需要在创建实体时设置创建人（对应Insert事件），只需给@before_event添加Insert参数，复用相同的上下文变量逻辑即可。
• 对于非请求触发的数据库操作（比如后台定时任务），需要手动调用current_user_var.set(xxx)设置用户，否则会触发无用户的错误。

内容的提问来源于stack exchange，提问作者andal_simmeth