FastAPI：保存接口返回JSON文件遇Product序列化报错求解决

场景复现 #

正常运行的代码 #

访问localhost:5000/products可正常获取JSON响应：

@router.get("/products", tags=["Product"], response_model=list[schemas.Product])
def get_all_product(db: Session = Depends(get_db)):
    return get_products(db)

def get_products(db: Session):
    products = db.query(models.Product).all()
    return products

修改后报错的代码 #

尝试将数据手动保存到本地文件时触发错误：

def get_all_product(db: Session = Depends(get_db)):
    with open('save_products.json', 'w', encoding = 'utf-8' ) as fp:
        json.dump(get_products(db), fp)
    return get_products(db)

错误信息 #

TypeError: Object of type Product is not JSON serializable.

原因解析 #

FastAPI能正常返回JSON是因为你指定了response_model=list[schemas.Product]，它会自动通过Pydantic模型将SQLAlchemy的Product实例转换成符合JSON格式的字典结构。而Python标准库的json.dump无法直接识别SQLAlchemy的模型实例——这是自定义对象，没有默认的序列化规则。

解决方案 #

方案1：使用FastAPI内置的jsonable_encoder #

FastAPI提供的jsonable_encoder可以直接把模型实例转换成可序列化的结构，和接口返回时的处理逻辑一致：

import json
from fastapi.encoders import jsonable_encoder

def get_all_product(db: Session = Depends(get_db)):
    products = get_products(db)
    # 转换为可JSON序列化的结构
    serialized_data = jsonable_encoder(products)
    with open('save_products.json', 'w', encoding='utf-8') as fp:
        json.dump(serialized_data, fp, ensure_ascii=False)
    return products

方案2：通过Pydantic模型手动转换 #

利用Pydantic的from_orm方法把SQLAlchemy实例转成Pydantic模型，再调用dict()方法得到字典：

import json

def get_all_product(db: Session = Depends(get_db)):
    products = get_products(db)
    # 逐个转换为Pydantic模型并转成字典
    product_dicts = [schemas.Product.from_orm(p).dict() for p in products]
    with open('save_products.json', 'w', encoding='utf-8') as fp:
        json.dump(product_dicts, fp, ensure_ascii=False)
    return products

方案3：给SQLAlchemy模型添加序列化方法 #

在models.Product类中添加一个to_dict方法，手动定义序列化规则：

# models.py中的Product类
class Product(Base):
    __tablename__ = "products"
    # 你的字段定义（比如id, name, price等）
    
    def to_dict(self):
        # 遍历表字段，生成字典
        return {column.name: getattr(self, column.name) for column in self.__table__.columns}

然后在保存时调用该方法：

import json

def get_all_product(db: Session = Depends(get_db)):
    products = get_products(db)
    product_dicts = [p.to_dict() for p in products]
    with open('save_products.json', 'w', encoding='utf-8') as fp:
        json.dump(product_dicts, fp, ensure_ascii=False)
    return products

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