如何用Pydantic+SQLAlchemy验证数据库唯一字段并在FastAPI中使用

下面是匹配你需求的完整实现，完全对齐原DRF的校验逻辑，同时适配FastAPI+Pydantic+SQLAlchemy的生态：

1. 前置依赖确认 #

假设你已经有这些基础组件（和原DRF代码对应）：

• SQLAlchemy的User模型，包含username和phone_number字段
• 密码复杂度校验函数check_password_complexity（不符合要求时抛出异常）
• 验证码服务ConfirmationService，其is_valid()方法校验失败时抛出异常
• 错误常量PHONE_NUMBER_IS_ALREADY_TAKEN和枚举ConfirmationCodeScopeChoice

2. 编写Pydantic校验模型 #

用Pydantic的模型级验证器模拟DRF的全局validate方法，同时支持传入SQLAlchemy Session做数据库校验：

from pydantic import BaseModel, model_validator, ValidationError
from sqlalchemy.orm import Session
from your_app.models import User
from your_app.utils import check_password_complexity
from your_app.services import ConfirmationService
from your_app.constants import PHONE_NUMBER_IS_ALREADY_TAKEN, ConfirmationCodeScopeChoice

class RegisterRequest(BaseModel):
    phone_number: str
    password: str
    code: str

    @model_validator(mode="after")
    def run_full_validation(self) -> "RegisterRequest":
        # 从模型上下文获取SQLAlchemy Session
        db: Session = self.model_config["context"]["db"]

        # 1. 校验手机号唯一性
        if db.query(User).filter(
            User.username == self.phone_number,
            User.phone_number == self.phone_number
        ).first():
            raise ValidationError(PHONE_NUMBER_IS_ALREADY_TAKEN)

        # 2. 校验密码复杂度
        check_password_complexity(self.password)

        # 3. 校验注册验证码
        ConfirmationService(
            phone_number=self.phone_number,
            scope=ConfirmationCodeScopeChoice.SIGN_UP,
            code=self.code
        ).is_valid()

        return self

3. FastAPI接口中的使用方式 #

方式一：手动校验（完全模拟DRF的is_valid()）

如果你想手动控制校验流程，和DRF里调用serializer.is_valid()的体验一致：

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from your_app.db import get_db  # 你的数据库Session依赖
from your_app.schemas import RegisterRequest
from pydantic import ValidationError

app = FastAPI()

@app.post("/register")
def register(request_data: dict, db: Session = Depends(get_db)):
    try:
        # 1. 实例化模型，先做基础字段校验（比如类型、非空）
        register_req = RegisterRequest(
            **request_data,
            model_config={"context": {"db": db}}
        )
        # 2. 模型实例化时会自动执行run_full_validation，校验不通过直接抛异常
        # 走到这里说明所有校验都通过了
    except ValidationError as e:
        # 返回和DRF类似的错误格式
        return {"detail": e.errors()}

    # 后续创建用户的业务逻辑
    # user = User(username=register_req.phone_number, phone_number=register_req.phone_number, ...)
    # db.add(user)
    # db.commit()
    return {"message": "注册成功"}

方式二：FastAPI自动校验（推荐）

利用FastAPI的自动请求解析和校验能力，代码更简洁：

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from your_app.db import get_db
from your_app.schemas import RegisterRequest

app = FastAPI()

@app.post("/register")
def register(
    register_req: RegisterRequest,
    db: Session = Depends(get_db)
):
    # 手动传入Session到模型上下文，执行自定义校验
    try:
        register_req.model_config["context"] = {"db": db}
        register_req.run_full_validation()
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=e.errors())

    # 后续业务逻辑
    return {"message": "注册成功"}

关键细节说明 #

• Pydantic的mode="after"验证器会在所有基础字段校验（比如手机号格式、密码非空）通过后再执行，和DRFvalidate方法的执行顺序一致。
• 必须显式传入SQLAlchemy Session，因为Pydantic模型本身不持有数据库连接，这是和DRF Serializer的核心区别（DRF依赖Django的ORM上下文）。
• 所有校验失败都会抛出ValidationError，可以统一捕获并返回标准化的错误响应，和DRF的serializer.errors逻辑对齐。

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