FastAPI嵌套Pydantic模型获取数据遇ValidationError问题

使用FastAPI、SQLAlchemy和Pydantic开发时，尝试从User、Company、UserPermissions三个关联模型中获取嵌套数据，已定义对应Pydantic Schema及路由接口。请求/info接口时返回18条ValidationError，提示每个用户的role字段下的order_read、item_write等字段缺失。但如果不使用UserPermissions相关Schema，仅返回role的UUID则正常，需排查错误原因。

模型代码 #

from sqlalchemy import Column, String, Boolean, TIMESTAMP, ForeignKey, text
from sqlalchemy.dialects.postgresql import UUID
import uuid
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(UUID(as_uuid=True), primary_key=True, nullable=False,
                default=uuid.uuid4)
    username = Column(String, nullable=False)
    email = Column(String, unique=True, nullable=False)
    phone = Column(String, unique=True, nullable=False)
    password = Column(String, nullable=False)
    photo = Column(String, nullable=True)
    company_id = Column(UUID(as_uuid=True), ForeignKey(
        'companies.id', ondelete='CASCADE'), nullable=True)
    verified = Column(Boolean, nullable=False, server_default='False')
    verification_code = Column(String, nullable=True, unique=True)
    # 仅定义外键，未建立关联关系
    role = Column(UUID(as_uuid=True), ForeignKey(
        'permissions.id', ondelete='CASCADE'), nullable=True)
    created_at = Column(TIMESTAMP(timezone=True),
                        nullable=False, server_default=text("now()"))
    updated_at = Column(TIMESTAMP(timezone=True),
                        nullable=False, server_default=text("now()"))

class Company(Base):
    __tablename__ = 'companies'
    id = Column(UUID(as_uuid=True), primary_key=True, nullable=False,
                default=uuid.uuid4)
    company_name = Column(String, nullable=False)
    inn = Column(String, nullable=True)
    email = Column(String, nullable=False)
    phone = Column(String(255), nullable=False)
    type = Column(String(12), nullable=False)
    created_at = Column(TIMESTAMP(timezone=True),
                        nullable=False, server_default=text("now()"))
    updated_at = Column(TIMESTAMP(timezone=True),
                        nullable=False, server_default=text("now()"))
    users = relationship('User')

class UserPermissions(Base):
    __tablename__ = 'permissions'
    id = Column(UUID(as_uuid=True), primary_key=True, nullable=False,
                default=uuid.uuid4)
    order_read = Column(Boolean, nullable=False, server_default='False')
    item_write = Column(Boolean, nullable=False, server_default='False')
    is_admin = Column(Boolean, nullable=False, server_default='False')
    permission_name = Column(String, nullable=False)
    created_at = Column(TIMESTAMP(timezone=True),
                        nullable=False, server_default=text("now()"))
    updated_at = Column(TIMESTAMP(timezone=True),
                        nullable=False, server_default=text("now()"))
    users = relationship('User')

Pydantic Schema代码 #

from pydantic import BaseModel, EmailStr
import uuid
from datetime import datetime
from typing import List, Optional

class CompanyBaseSchema(BaseModel):
    company_name: str
    email: EmailStr
    phone: str
    inn: Optional[str] = None
    type: str

    class Config:
        orm_mode = True
    
class CompanyInfoResponse(BaseModel):
    status: str
    users_amount: int
    company: "CompanyUsersResponse"

class CompanyUsersResponse(CompanyBaseSchema):
    id: uuid.UUID
    created_at: datetime
    inn: Optional[str] = None
    users: List["FilteredUserResponse"]
    updated_at: datetime

class PermissionBaseSchema(BaseModel):
    order_read: bool
    item_write: bool
    is_admin: bool
    permission_name: str
    created_at: datetime
    updated_at: datetime

    class Config:
        orm_mode = True

class FilteredPermissionSchema(PermissionBaseSchema):
    id: Optional[uuid.UUID] = None

class UserBaseSchema(BaseModel):
    username: str
    email: EmailStr
    phone: str
    company_id: Optional[uuid.UUID] = None
    role: uuid.UUID

    class Config:
        orm_mode = True

class FilteredUserResponse(UserBaseSchema):
    username: Optional[str] = None
    email: Optional[EmailStr] = None
    phone: Optional[str] = None
    role: FilteredPermissionSchema

# 解决前向引用问题
CompanyInfoResponse.update_forward_refs()
CompanyUsersResponse.update_forward_refs()

路由代码 #

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from . import schemas, models
from .dependencies import get_db, require_user

router = APIRouter()

@router.get('/info', response_model=schemas.CompanyInfoResponse)
def get_company_info(db: Session = Depends(get_db),
                     user_id: str = Depends(require_user)):
    user_company_one = db.query(models.User).filter(models.User.id == user_id).first().company_id
    company_info = get_post(str(user_company_one), db, user_id) # 自定义查询Company的函数
    return {'status': 'success', 'users_amount': len(company_info.users), 'company': company_info}

错误信息 #

pydantic.error_wrappers.ValidationError: 18 validation errors for CompanyInfoResponse
response -> company -> users -> 0 -> role -> order_read
  field required (type=value_error.missing)
response -> company -> users -> 0 -> role -> item_write
  field required (type=value_error.missing)
response -> company -> users -> 0 -> role -> is_admin
  field required (type=value_error.missing)
response -> company -> users -> 0 -> role -> permission_name
  field required (type=value_error.missing)
response -> company -> users -> 0 -> role -> created_at
  field required (type=value_error.missing)
response -> company -> users -> 0 -> role -> updated_at
  field required (type=value_error.missing)
response -> company -> users -> 1 -> role -> order_read
  field required (type=value_error.missing)
response -> company -> users -> 1 -> role -> item_write
  field required (type=value_error.missing)
response -> company -> users -> 1 -> role -> is_admin
  field required (type=value_error.missing)
response -> company -> users -> 1 -> role -> permission_name
  field required (type=value_error.missing)
response -> company -> users -> 1 -> role -> created_at
  field required (type=value_error.missing)
response -> company -> users -> 1 -> role -> updated_at
  field required (type=value_error.missing)
response -> company -> users -> 2 -> role -> order_read
  field required (type=value_error.missing)
response -> company -> users -> 2 -> role -> item_write
  field required (type=value_error.missing)
response -> company -> users -> 2 -> role -> is_admin
  field required (type=value_error.missing)
response -> company -> users -> 2 -> role -> permission_name
  field required (type=value_error.missing)
response -> company -> users -> 2 -> role -> created_at
  field required (type=value_error.missing)
response -> company -> users -> 2 -> role -> updated_at
  field required (type=value_error.missing)

错误原因 #

1. SQLAlchemy关联关系缺失：User模型中的role仅为外键字段，未与UserPermissions建立relationship关联。查询User实例时，SQLAlchemy不会自动加载对应的权限数据，返回的role只是UUID值，而非完整的UserPermissions对象。
2. Pydantic Schema不匹配：FilteredUserResponse中定义role为FilteredPermissionSchema，要求传入包含order_read等字段的完整权限对象，但实际从数据库获取的User实例里role仅为UUID，缺少权限字段，导致校验失败。

解决方案 #

步骤1：完善SQLAlchemy模型关联 #

修改User模型，将外键字段与关联属性分离，建立与UserPermissions的关联：

class User(Base):
    __tablename__ = 'users'
    # 保留其他字段不变
    # 重命名外键字段为role_id
    role_id = Column(UUID(as_uuid=True), ForeignKey(
        'permissions.id', ondelete='CASCADE'), nullable=True)
    # 建立relationship关联，获取完整权限对象
    role = relationship('UserPermissions')

步骤2：预加载嵌套关联数据 #

在查询Company的get_post函数中，使用SQLAlchemy的joinedload预加载用户及对应权限数据，确保一次性获取所有嵌套数据，同时避免N+1查询问题：

from sqlalchemy.orm import joinedload

def get_post(company_id: str, db: Session, user_id: str):
    return db.query(models.Company).options(
        joinedload(models.Company.users).joinedload(models.User.role)
    ).filter(models.Company.id == company_id).first()

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