FastAPI中使用Pydantic进行请求响应模型校验时出错

在POST请求中使用不同的输入输出模型时，遇到字段校验错误，以下是相关代码及报错信息：

路由代码 #

@router.post('/sign-up/', response_model=UserOutSchema)
def register(user: UserInSchema, session: AsyncSession = Depends(get_async_session)):
    hashed_password = bcrypt.hashpw(user.password1.encode('utf-8'), bcrypt.gensalt())
    db_user = UserModel(
        name=user.name, surname=user.surname, email=user.email, hashed_password=hashed_password
    )
    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user

Schema定义 #

class UserIn(BaseModel):
    name: str
    surname: str
    email: EmailStr
    password1: str
    password2: str

    @model_validator(mode='after')
    def check_passwords_match(self):
        pw1 = self.password1
        pw2 = self.password2

        if pw1 is not None and pw2 is not None and pw1 != pw2:
            raise ValueError('密码不匹配')
        return self

    @classmethod
    @field_validator('email')
    def validate_email_taken(cls):
        email = cls.email
        query = db.query(UserModel).filter(UserModel.email == email).first()
        if query:
            raise ValueError('邮箱已存在')
        return cls


class UserOut(BaseModel):
    id: int
    name: str
    surname: str
    email: EmailStr
    time_created: datetime
    email_confirmed: bool
    is_admin: bool
    is_stuff: bool

注：UserOutSchema是UserOut的导入别名

数据库模型 #

class AbstractModel(DeclarativeBase):
    id: Mapped[int] = mapped_column(primary_key=True, index=True, autoincrement=True)


class User(AbstractModel):
    __tablename__ = 'users'

    name: Mapped[str] = mapped_column(String(50), nullable=False)
    surname: Mapped[str] = mapped_column(String(50), nullable=False)
    email: Mapped[str] = mapped_column(String(70), unique=True, index=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String(50), nullable=False)

    time_created = mapped_column(DateTime(timezone=True), server_default=func.now())

    email_confirmed: Mapped[bool] = mapped_column(Boolean, default=False)
    is_seller: Mapped[bool] = mapped_column(default=False)
    is_admin: Mapped[bool] = mapped_column(default=False)
    is_stuff: Mapped[bool] = mapped_column(default=False)

    seller: Mapped['Seller'] = relationship(back_populates='user', uselist=False)

    def __repr__(self) -> str:
        return f'User<{self.id}> {self.name} + {self.surname}'


class Seller(AbstractModel):
    __tablename__ = 'sellers'

    user: Mapped['User'] = relationship(back_populates='seller', uselist=False)
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))

    mobile: Mapped[str] = mapped_column(String(30), unique=True, nullable=False)
    company_name: Mapped[str] = mapped_column(unique=True, nullable=False)
    company_description: Mapped[str] = mapped_column(Text, nullable=False)

    bank_name: Mapped[str] = mapped_column(nullable=False)
    tin: Mapped[str] = mapped_column(String(10), unique=True, nullable=False)     # 纳税人识别号
    bic: Mapped[str] = mapped_column(String(9), nullable=False)           # 银行识别码
    trrc: Mapped[str] = mapped_column(String(9), nullable=False)          # 税务登记原因码
    an: Mapped[str] = mapped_column(String(20), nullable=False, unique=True)           # 账号

    passport_scan: Mapped[str] = mapped_column(String(130), nullable=False, unique=True)
    tc_scan: Mapped[str] = mapped_column(String(130), nullable=False, unique=True)       # 商标证书扫描件

    additional: Mapped[str | None] = mapped_column(Text, nullable=True)
    time_created = mapped_column(DateTime(timezone=True), server_default=func.now())

    # 审核通过后可售卖商品
    is_approved: Mapped[bool] = mapped_column(default=False)

    def __repr__(self):
        return f'User<{self.user_id}> Seller<{self.id}>'

报错栈信息 #

weatherforecast-web-1      | RuntimeWarning: Enable tracemalloc to get the object allocation traceback
weatherforecast-web-1      | INFO:     172.20.0.1:51818 - "POST /users/sign-up/ HTTP/1.1" 500 Internal Server Error
weatherforecast-web-1      | ERROR:    Exception in ASGI application
weatherforecast-web-1      | Traceback (most recent call last):
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/uvicorn/protocols/http/httptools_impl.py", line 426, in run_asgi
weatherforecast-web-1      |     result = await app(  # type: ignore[func-returns-value]
weatherforecast-web-1      |              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/uvicorn/middleware/proxy_headers.py", line 84, in __call__
weatherforecast-web-1      |     return await self.app(scope, receive, send)
weatherforecast-web-1      |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/fastapi/applications.py", line 1115, in __call__
weatherforecast-web-1      |     await super().__call__(scope, receive, send)
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/starlette/applications.py", line 122, in __call__
weatherforecast-web-1      |     await self.middleware_stack(scope, receive, send)
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/starlette/middleware/errors.py", line 184, in __call__
weatherforecast-web-1      |     raise exc
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/starlette/middleware/errors.py", line 162, in __call__
weatherforecast-web-1      |     await self.app(scope, receive, _send)
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/starlette/middleware/exceptions.py", line 79, in __call__
weatherforecast-web-1      |     raise exc
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/starlette/middleware/exceptions.py", line 68, in __call__
weatherforecast-web-1      |     await self.app(scope, receive, sender)
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/fastapi/middleware/asyncexitstack.py", line 20, in __call__
weatherforecast-web-1      |     raise e
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/fastapi/middleware/asyncexitstack.py", line 17, in __call__
weatherforecast-web-1      |     await self.app(scope, receive, send)
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/starlette/routing.py", line 718, in __call__
weatherforecast-web-1      |     await route.handle(scope, receive, send)
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/starlette/routing.py", line 276, in handle
weatherforecast-web-1      |     await self.app(scope, receive, send)
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/starlette/routing.py", line 66, in app
weatherforecast-web-1      |     response = await func(request)
weatherforecast-web-1      |                ^^^^^^^^^^^^^^^^^^^
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/fastapi/routing.py", line 292, in app
weatherforecast-web-1      |     content = await serialize_response(
weatherforecast-web-1      |               ^^^^^^^^^^^^^^^^^^^^^^^^^
weatherforecast-web-1      |   File "/usr/local/lib/python3.12/site-packages/fastapi/routing.py", line 155, in serialize_response
weatherforecast-web-1      |     raise ResponseValidationError(
weatherforecast-web-1      | fastapi.exceptions.ResponseValidationError: 5个校验错误：
weatherforecast-web-1      |   {'type': 'int_type', 'loc': ('response', 'id'), 'msg': '输入应为有效整数', 'input': None}
weatherforecast-web-1      |   {'type': 'datetime_type', 'loc': ('response', 'time_created'), 'msg': '输入应为有效日期时间', 'input': None}
weatherforecast-web-1      |   {'type': 'bool_type', 'loc': ('response', 'email_confirmed'), 'msg': '输入应为有效布尔值', 'input': None}
weatherforecast-web-1      |   {'type': 'bool_type', 'loc': ('response', 'is_admin'), 'msg': '输入应为有效布尔值', 'input': None}
weatherforecast-web-1      |   {'type': 'bool_type', 'loc': ('response', 'is_stuff'), 'msg': '输入应为有效布尔值', 'input': None}

错误原因 #

1. 异步Session与同步路由不兼容：路由函数是同步def，但依赖了AsyncSession，导致session.commit()和session.refresh()无法异步执行，数据库生成的id、time_created等字段没有被刷新到db_user对象中，返回时这些字段为None，触发Pydantic校验失败。
2. 邮箱校验函数写法错误：@field_validator的类方法未接收字段值参数，且使用同步查询方式，无法适配异步Session。
3. 密码哈希值类型不匹配：bcrypt生成的哈希值是bytes类型，直接存入数据库的String字段会导致类型异常，且可能影响后续认证逻辑。

修复步骤 #

1. 将路由改为异步函数 #

同步路由无法正确使用AsyncSession，需改为异步函数，并对数据库操作添加await：

@router.post('/sign-up/', response_model=UserOutSchema)
async def register(user: UserInSchema, session: AsyncSession = Depends(get_async_session)):
    # 将bytes类型的哈希值解码为字符串
    hashed_password = bcrypt.hashpw(user.password1.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
    # 注意：这里的User是数据库模型类，若原代码用了UserModel别名，需保持一致
    db_user = User(
        name=user.name, surname=user.surname, email=user.email, hashed_password=hashed_password
    )
    session.add(db_user)
    await session.commit()
    await session.refresh(db_user)
    return db_user

2. 修复邮箱唯一性校验 #

Pydantic同步校验函数无法使用异步Session，建议将邮箱校验移到FastAPI依赖项中：

# 移除UserIn类中的同步邮箱校验函数
class UserIn(BaseModel):
    name: str
    surname: str
    email: EmailStr
    password1: str
    password2: str

    @model_validator(mode='after')
    def check_passwords_match(self):
        pw1 = self.password1
        pw2 = self.password2

        if pw1 is not None and pw2 is not None and pw1 != pw2:
            raise ValueError('密码不匹配')
        return self

# 定义异步依赖项校验邮箱是否存在
from sqlalchemy import select
from fastapi import HTTPException, Depends

async def check_email_exists(email: str, session: AsyncSession = Depends(get_async_session)):
    existing_user = await session.execute(select(User).where(User.email == email))
    if existing_user.scalar_one_or_none():
        raise HTTPException(status_code=400, detail='邮箱已存在')

# 修改路由，添加该依赖
@router.post('/sign-up/', response_model=UserOutSchema)
async def register(
    user: UserInSchema, 
    session: AsyncSession = Depends(get_async_session),
    _: None = Depends(check_email_exists)
):
    hashed_password = bcrypt.hashpw(user.password1.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
    db_user = User(
        name=user.name, surname=user.surname, email=user.email, hashed_password=hashed_password
    )
    session.add(db_user)
    await session.commit()
    await session.refresh(db_user)
    return db_user

3. 规范数据库模型字段定义 #

为time_created添加Mapped类型注解，保持与其他字段一致：

class User(AbstractModel):
    __tablename__ = 'users'
    # ... 其他字段
    time_created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    # ... 其他字段

4. 确认类别名一致性 #

确保路由中使用的UserModel是数据库模型User的正确导入别名，避免实例化错误。

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