如何在FastAPI的Pydantic模型中复用请求级SQLAlchemy会话做校验

场景背景 #

基于FastAPI、Pydantic和SQLAlchemy栈开发API时，遇到需要通过数据库查询校验请求体参数的场景。以/forgot-password接口为例：

• 接口接收email参数，需校验邮箱是否存在于数据库
• 存在则执行生成令牌、发送邮件等后续操作
• 不存在则要求Pydantic抛出标准ValidationError，保证校验错误响应格式统一，便于客户端处理

请求体示例：

{
    "email": "example@gmail.com"
}

现有实现的问题 #

初始实现：无法访问请求会话 #

尝试在Pydantic字段校验器中查询数据库，但无法获取请求上下文的活跃SQLAlchemy会话：

class ForgotPasswordRequestSchema(BaseModel):
    email: EmailStr
    
    @validator("email")
    def validate_email(cls, v):
        # 此处无法访问请求级别的活跃会话
        user = session.get(Users, email=v) 
        if not user:
            raise ValueError("Email does not exist in the database.")

        return v

直接创建会话：会话管理矛盾 #

若在模型中直接创建会话，会引发会话生命周期管理问题：

class ForgotPasswordRequestSchema(BaseModel):
    email: EmailStr
    _session = get_db() # 返回数据库会话
    _user = None
    
    @validator("email")
    def validate_email(cls, v):
        user = cls._session.get(Users, email=v)

        if not user:
            cls._session.close()
            raise ValueError("Email does not exist in the database.")

        cls._user = user
        cls._session.close()

        return v

• 若关闭会话：后续接口处理函数无法复用已查询的_user对象，需重复查询数据库
• 若不关闭会话：会触发sqlalchemy.exc.PendingRollbackError等异常，违背"一个请求周期使用一个会话"的原则

需求 #

需要将FastAPI依赖提供的请求级会话传递给Pydantic模型，复用同一会话完成参数校验与后续业务处理，避免重复查询和会话管理问题。

解决方案：通过FastAPI依赖注入传递会话到Pydantic验证 #

利用Pydantic的模型级验证器结合FastAPI的依赖注入机制，实现会话的传递与复用，同时保证校验错误响应的一致性。

1. 定义带模型级验证的Pydantic模型 #

使用Pydantic v2的model_validator(mode='after')（v1版本可使用@root_validator(skip_on_failure=True)），预留会话注入的属性：

from pydantic import BaseModel, EmailStr, model_validator
from sqlalchemy.orm import Session
from your_db_module import Users

class ForgotPasswordRequestSchema(BaseModel):
    email: EmailStr
    _user: Users | None = None
    _db_session: Session | None = None  # 用于注入请求级会话

    @model_validator(mode='after')
    def validate_email_and_fetch_user(self):
        if not self._db_session:
            raise ValueError("Database session not provided")
        
        # 使用注入的会话查询用户
        user = self._db_session.query(Users).filter(Users.email == self.email).first()
        if not user:
            raise ValueError("Email does not exist in the database.")
        
        self._user = user  # 缓存用户对象供后续使用
        return self

2. 自定义依赖封装验证逻辑 #

创建自定义依赖，结合会话依赖与模型验证，将校验后的模型实例传递给路由处理函数：

from fastapi import Depends, FastAPI
from pydantic import ValidationError
from your_db_module import get_db

app = FastAPI()

def validate_forgot_password_request(
    request_data: dict,
    db: Session = Depends(get_db)
) -> ForgotPasswordRequestSchema:
    try:
        # 实例化模型并注入请求级会话
        schema = ForgotPasswordRequestSchema(**request_data)
        schema._db_session = db
        # 触发模型验证
        schema.validate_email_and_fetch_user()
        return schema
    except ValueError as e:
        # 将自定义错误转换为Pydantic标准ValidationError，保证响应格式统一
        raise ValidationError([
            {
                "loc": ("email",),
                "msg": str(e),
                "type": "value_error.email_not_found"
            }
        ])

3. 路由中使用自定义依赖 #

@app.post("/forgot-password")
async def forgot_password(
    validated_request: ForgotPasswordRequestSchema = Depends(validate_forgot_password_request)
):
    # 直接复用已查询的用户对象，无需重复查询
    user = validated_request._user
    
    # 执行生成令牌、发送邮件等操作
    # generate_reset_token(user)
    # send_reset_email(user.email, token)
    
    return {"status": "success", "message": "Password reset email sent"}

方案优势 #

• 复用请求级SQLAlchemy会话，符合会话生命周期管理最佳实践
• 校验逻辑封装在Pydantic模型中，保证错误响应格式统一（FastAPI会自动将ValidationError转换为422状态码的标准响应）
• 后续业务处理直接使用已缓存的用户对象，避免重复数据库查询

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