FastAPI中Pydantic2与SQLAlchemy嵌套数据序列化问题排查

在基于FastAPI、PostgreSQL、SQLAlchemy和Pydantic2的项目中，已建立User与Profile的一对一关联，尝试通过Pydantic模型返回包含关联User的Profile数据时，抛出ValidationError，提示User对象无法转换为ReadUser类型。

相关代码 #

User模型 #

class User(Base):
    __tablename__ = "users"
    id = Column(
        UUID(as_uuid=True),
        nullable=False,
        primary_key=True,
        default=uuid.uuid4,
    )
    ...<some other fields>
    profile = relationship("Profile", back_populates="user")

Profile模型 #

class Profile(Base):
    __tablename__ = "profiles"

    id = Column(
        UUID(as_uuid=True),
        nullable=False,
        primary_key=True,
        default=uuid.uuid4,
    )
    user_id = Column(
        UUID(as_uuid=True),
        ForeignKey("users.id"),
        nullable=False,
        unique=True,
    )
    user = relationship(
        "User",
        foreign_keys="Profile.user_id",
        back_populates="profile"
    )
    ...<some other fields>

获取Profile的路由 #

@router.get(
    "{user_id}",
    status_code=status.HTTP_200_OK,
    response_model=ReadProfile,
)
async def get_profile_read(
    profile_id: str,
    profile_service: ProfilesService = Depends(get_profiles_service),
    current_user: str = Depends(get_auth),
):
    profile = await profile_service.get(profile_id)
    return ReadProfile(**profile.__dict__)

Pydantic模型 #

class ReadProfile(BaseModel):
    id: UUID
    ...<some other data>
    user: ReadUser

class ReadUser(BaseModel):
    id: UUID
    ...some other data

Service层获取Profile方法 #

async def get(self, profile_id: str) -> ReadProfile:
    q = await self.db_session.execute(
        select(Profile)
        .where(Profile.id == profile_id)
        .options(
            selectinload(Profile.user)
        )
    )
    result = q.scalars().first()
    if not result:
        raise ValueError("Profile not found!")
    return result

错误信息 #

pydantic_core._pydantic_core.ValidationError: 1 validation error for ReadProfile
web_1      | user
web_1      |   Input should be a valid dictionary or instance of ReadUser [type=model_type, input_value=<app.models.user.User object at 0x7f8b7c3cca30>, input_type=User]

错误原因 #

直接将SQLAlchemy的User对象传递给Pydantic的ReadUser字段时，Pydantic无法自动识别并转换SQLAlchemy模型实例，即使通过__dict__传递，user字段仍然是SQLAlchemy的User实例，而非Pydantic期望的字典或ReadUser实例。

解决方法 #

方法1：使用Pydantic的from_orm方法 #

Pydantic 2.x支持直接从ORM模型实例创建Pydantic对象，会自动递归处理关联关系：

# 修改路由中的返回逻辑
return ReadProfile.from_orm(profile)

方法2：手动转换关联对象 #

如果坚持使用__dict__，需手动将user字段转换为ReadUser实例：

profile_dict = profile.__dict__.copy()
# 移除SQLAlchemy内部状态属性（可选但推荐）
profile_dict.pop('_sa_instance_state', None)
# 转换关联的User对象
profile_dict['user'] = ReadUser.from_orm(profile.user)
return ReadProfile(**profile_dict)

方法3：配置Pydantic模型的ORM模式（推荐） #

在Pydantic模型中添加model_config = ConfigDict(from_attributes=True)，FastAPI会自动将ORM模型转换为Pydantic模型，无需手动实例化：

class ReadUser(BaseModel):
    id: UUID
    ...<some other data>
    model_config = ConfigDict(from_attributes=True)

class ReadProfile(BaseModel):
    id: UUID
    ...<some other data>
    user: ReadUser
    model_config = ConfigDict(from_attributes=True)

之后路由可直接返回SQLAlchemy实例：

async def get_profile_read(
    profile_id: str,
    profile_service: ProfilesService = Depends(get_profiles_service),
    current_user: str = Depends(get_auth),
):
    profile = await profile_service.get(profile_id)
    return profile  # FastAPI自动用response_model转换

额外修正：路由参数不匹配问题 #

路由路径参数是{user_id}，但函数参数是profile_id，需修正为一致：

@router.get(
    "{profile_id}",  # 改为profile_id
    status_code=status.HTTP_200_OK,
    response_model=ReadProfile,
)
async def get_profile_read(
    profile_id: str,
    # ... 其他参数
):

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