FastAPI如何为每个用户实现自定义Pydantic Schema？

一、修正动态导入方案 #

动态导入完全可行，用Python标准库importlib就能实现，之前没成功大概率是实现方式不对。步骤如下：

1. 从请求中解析出用户标识（比如从Bearer Token提取用户名）
2. 动态加载对应用户的Schema模块
3. 获取模块内的自定义Pydantic模型

示例代码：

import importlib
from fastapi import FastAPI, Request, Depends, HTTPException
from pydantic import BaseModel

# 假设auth模块提供解析token获取用户名的函数
from auth import get_current_username

app = FastAPI()

def get_user_model(username: str = Depends(get_current_username)):
    try:
        # 动态导入用户模块
        module = importlib.import_module(f"schemas.{username}.schema")
        # 获取模块中的自定义PydanticClass
        return getattr(module, "PydanticClass")
    except (ImportError, AttributeError):
        # 用户无自定义模型时，回退到base模型
        from schemas.base.schema import PydanticClass as BaseModel
        return BaseModel

@app.post("/create")
async def create_something(request: Request, UserModel = Depends(get_user_model)):
    # 用对应用户的模型解析请求体
    raw_data = await request.json()
    parsed_data = UserModel(**raw_data)
    # 后续业务逻辑处理
    return {"processed_data": parsed_data.dict()}

二、更贴合FastAPI风格的依赖注入方案 #

上面的方法需要手动解析请求体，也可以通过自定义依赖让FastAPI自动完成解析，代码更简洁：

import importlib
from fastapi import FastAPI, Depends
from pydantic import BaseModel
from typing import Annotated, Type

from auth import get_current_username

app = FastAPI()

def get_user_model(username: str = Depends(get_current_username)) -> Type[BaseModel]:
    try:
        module = importlib.import_module(f"schemas.{username}.schema")
        return getattr(module, "PydanticClass")
    except (ImportError, AttributeError):
        from schemas.base.schema import PydanticClass
        return PydanticClass

def parse_request_body(model: Type[BaseModel] = Depends(get_user_model)):
    async def _parse(request):
        raw_data = await request.json()
        return model(**raw_data)
    return _parse

@app.post("/create")
async def create_something(parsed_data: Annotated[BaseModel, Depends(parse_request_body)]):
    # parsed_data已经是对应用户模型的实例，直接使用即可
    return {"result": parsed_data.dict()}

三、备选思路：模型注册字典 #

如果不想用动态导入，可提前把所有用户模型注册到字典中，根据用户名直接获取，适合用户数量较少的场景：

from fastapi import FastAPI, Request, Depends
from pydantic import BaseModel

from auth import get_current_username
from schemas.base.schema import PydanticClass as BaseModel
from schemas.user_A.schema import PydanticClass as UserAModel
from schemas.user_B.schema import PydanticClass as UserBModel

# 提前注册所有用户模型
USER_MODEL_MAP = {
    "user_A": UserAModel,
    "user_B": UserBModel
}

app = FastAPI()

def get_user_model(username: str = Depends(get_current_username)) -> Type[BaseModel]:
    return USER_MODEL_MAP.get(username, BaseModel)

@app.post("/create")
async def create_something(request: Request, model = Depends(get_user_model)):
    raw_data = await request.json()
    parsed_data = model(**raw_data)
    return {"data": parsed_data.dict()}

关键注意事项 #

• 对获取到的用户名做合法性校验，比如限制只能是字母、下划线组合，防止目录遍历攻击
• 必须处理模型导入失败的情况，要么回退到base模型，要么返回明确的错误响应
• 动态导入的模块会被Python缓存，修改用户自定义模型后需要重启服务

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