FastAPI+SQLAlchemy跨表查询返回结果触发TypeError问题咨询

问题背景 #

使用FastAPI查询Post表与Likes表，通过关联查询获取Post信息及对应点赞数，代码执行触发500内部服务器错误，报错显示序列化失败。打印结果发现返回的是元组列表（[(<Post对象>, 0), ...]）而非预期的字典列表，已通过列表推导式临时解决，但疑惑为何教程代码可正常运行，寻求更优方案。

原代码 #

from fastapi import APIRouter, status, Depends
from sqlalchemy.orm import Session
from sqlalchemy import func
import models

router = APIRouter()

@router.get('/', status_code= status.HTTP_200_OK)
def get_notes(db: Session = Depends(get_db)):
    posts = db.query(models.Post, func.count(models.Likes.post_id).label("likes")) \
.join(models.Likes, models.Likes.post_id == models.Post.post_id, isouter=True) \
.group_by(models.Post.post_id).all()
    return posts

报错信息 #

File "C:\Users\apapa\AppData\Local\pypoetry\Cache\virtualenvs\fastapi-backend-curI0rdX-py3.10\lib\site-packages\fastapi\encoders.py", line 160, in jsonable_encoder
    raise ValueError(errors) from e
ValueError: [TypeError('cannot convert dictionary update sequence element #0 to a sequence'), TypeError('vars() argument must have __dict__ attribute')]

临时解决代码 #

posts = [{"Post": {**post.__dict__}, "likes": likes} for post, likes in posts]
return posts

原因分析 #

序列化失败核心原因 #

FastAPI的jsonable_encoder无法直接序列化SQLAlchemy模型对象与元组的组合。查询返回的是(Post实例, 点赞数)的元组列表，默认编码器既不知道如何将SQLAlchemy实例转为可序列化字典，也无法处理元组的结构转换。

教程代码可正常运行的可能原因 #

1. 模型配置差异：教程中的SQLAlchemy模型可能继承了自定义基类，该基类实现了序列化优化（比如自动排除_sa_instance_state这类无用属性）。
2. 版本差异：早期版本的FastAPI对SQLAlchemy实例的序列化兼容性更好，或教程使用的SQLAlchemy版本默认返回可直接序列化的结构。
3. 遗漏Pydantic包装：教程中可能用Pydantic响应模型包装了查询结果，但你复制代码时遗漏了这一步。

更优解决方案 #

方案1：使用Pydantic响应模型（推荐） #

定义对应的Pydantic模型，让FastAPI自动处理序列化，同时保证数据结构合法性：

from pydantic import BaseModel
from datetime import datetime

# Post基础模型
class PostBase(BaseModel):
    post_id: int
    user_id: int
    title: str
    content: str
    created_datetime: datetime

    class Config:
        orm_mode = True  # 允许从SQLAlchemy实例创建模型

# 包含点赞数的响应模型
class PostWithLikes(BaseModel):
    Post: PostBase
    likes: int

# 修改接口指定响应模型
@router.get('/', status_code=status.HTTP_200_OK, response_model=list[PostWithLikes])
def get_notes(db: Session = Depends(get_db)):
    posts = db.query(models.Post, func.count(models.Likes.post_id).label("likes")) \
    .join(models.Likes, models.Likes.post_id == models.Post.post_id, isouter=True) \
    .group_by(models.Post.post_id).all()
    return [{"Post": post, "likes": likes} for post, likes in posts]

此方案利用Pydantic的orm_mode自动转换SQLAlchemy实例，同时提供类型校验和自动生成接口文档。

方案2：让SQLAlchemy直接返回字典结构 #

通过with_entities指定返回字段，将查询结果直接转为字典：

@router.get('/', status_code=status.HTTP_200_OK)
def get_notes(db: Session = Depends(get_db)):
    posts = db.query(
        models.Post.post_id.label("post_id"),
        models.Post.user_id.label("user_id"),
        models.Post.title.label("title"),
        models.Post.content.label("content"),
        models.Post.created_datetime.label("created_datetime"),
        func.count(models.Likes.post_id).label("likes")
    ).join(models.Likes, models.Likes.post_id == models.Post.post_id, isouter=True) \
    .group_by(models.Post.post_id).all()
    # 将SQLAlchemy Row对象转为字典
    return [dict(post._mapping) for post in posts]

这种方式直接返回包含所有字段的字典，无需处理模型实例，但需要手动指定所有字段。

方案3：优化临时解决方法 #

避免直接使用__dict__（包含_sa_instance_state等无用属性），用SQLAlchemy工具方法提取有效字段：

from sqlalchemy.orm import class_mapper

def serialize_sqlalchemy_obj(obj):
    """序列化SQLAlchemy模型实例为字典"""
    return {column.key: getattr(obj, column.key) for column in class_mapper(obj.__class__).columns}

# 修改临时代码
posts = [{"Post": serialize_sqlalchemy_obj(post), "likes": likes} for post, likes in posts]
return posts

此方法仅提取模型中定义的数据库字段，避免冗余属性。

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