FastAPI+SQLAlchemy联表查询时vars()序列化错误求助

错误原因 #

你的查询返回的是元组列表，每个元组包含SQLAlchemy的Product模型实例和votes数值。FastAPI的jsonable_encoder无法直接序列化这种混合结构：元组本身没有__dict__属性，默认序列化逻辑无法正确处理元组内模型实例与数值的组合，因此抛出vars()相关错误。

解决方法 #

方法一：手动转换为可序列化字典 #

遍历查询结果，将每个元组转换成符合期望格式的字典，同时清理SQLAlchemy模型的内部属性：

@router.get("/")
def get_products(db: Session = Depends(get_db)):
    products = db.query(models.Product, func.count(models.Vote.product_id).label("votes")).join(
        models.Vote, models.Vote.product_id == models.Product.id, isouter=True).group_by(models.Product.id).all()
    
    result = []
    for product, votes in products:
        # 提取Product字段，排除SQLAlchemy内部属性
        product_dict = {k: v for k, v in product.__dict__.items() if not k.startswith('_')}
        result.append({
            "Product": product_dict,
            "votes": votes
        })
    return result

方法二：使用SQLAlchemy inspect工具转换模型 #

利用SQLAlchemy的inspect方法获取模型的持久化属性，避免手动过滤内部字段：

from sqlalchemy import inspect

@router.get("/")
def get_products(db: Session = Depends(get_db)):
    products = db.query(models.Product, func.count(models.Vote.product_id).label("votes")).join(
        models.Vote, models.Vote.product_id == models.Product.id, isouter=True).group_by(models.Product.id).all()
    
    result = []
    for product, votes in products:
        # 获取模型的属性字典并提取值
        product_attrs = inspect(product).attrs
        clean_product = {k: v.value for k, v in product_attrs.items()}
        result.append({
            "Product": clean_product,
            "votes": votes
        })
    return result

方法三：使用Pydantic模型定义返回结构（推荐） #

创建匹配期望输出的Pydantic模型，利用其ORM序列化能力自动处理转换：

首先定义Pydantic模型：

from pydantic import BaseModel
from datetime import datetime

class ProductResponse(BaseModel):
    id: int
    is_available: bool
    user_id: int
    title: str
    price: float
    created_at: datetime

    class Config:
        orm_mode = True  # 支持从SQLAlchemy模型实例初始化

class ProductWithVotes(BaseModel):
    Product: ProductResponse
    votes: int

然后修改接口：

@router.get("/", response_model=list[ProductWithVotes])
def get_products(db: Session = Depends(get_db)):
    products = db.query(models.Product, func.count(models.Vote.product_id).label("votes")).join(
        models.Vote, models.Vote.product_id == models.Product.id, isouter=True).group_by(models.Product.id).all()
    
    return [ProductWithVotes(Product=product, votes=votes) for product, votes in products]

这种方法符合FastAPI最佳实践，自动生成接口文档，同时保证序列化的正确性。

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