FastAPI遇PydanticSerializationError：无法序列化SQLAlchemy模型求助

错误信息 #

pydantic_core._pydantic_core.PydanticSerializationError: Unable to serialize unknown type: <class 'model.standard_data_model.Standard_Data_Model'>

相关代码 #

SQLAlchemy模型定义 #

from sqlalchemy import Column, INT, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
import datetime

class Standard_Data_Model(declarative_base()):
    __tablename__ = 'miun_standard_data'
    id = Column(INT, primary_key=True, autoincrement=True, comment='Primary key id')
    ros_bag_id = Column(INT, nullable=True, comment='Associated raw data set management')
    language_instruction = Column(String(255), nullable=True, comment='Instruction description')
    duration = Column(String(255), nullable=True, comment='Duration')
    morphology = Column(String(255), nullable=True, comment='Morphology')
    robot_type = Column(String(255), nullable=True, comment='Device type')
    effector_type = Column(String(255), nullable=True, comment='Gripping type')
    device_number = Column(String(255), nullable=True, comment='Device number')
    resource_type = Column(String(255), nullable=True, comment='File type')
    file_path = Column(String(255), nullable=True, comment='File path')
    status = Column(String(255), nullable=True, comment='Status')
    task_id = Column(String(255), nullable=True, comment='Task id')
    editor = Column(String(255), nullable=True, comment='Editor id')
    description = Column(String(255), nullable=True, comment='Description of this message')
    tag = Column(String(255), nullable=True, comment='Tag set')
    version_num = Column(String(255), nullable=True, comment='Version number')
    create_time = Column(DateTime, default=datetime.now(), comment='creation time')
    update_time = Column(DateTime, default=None, onupdate=datetime.now(), comment='update time')

    def __init__(self, ros_bag_id, language_instruction, duration, morphology, robot_type,
                 effector_type, device_number, resource_type, file_path, status, task_id,
                 editor, description, tag, version_num, create_time, update_time):
        self.ros_bag_id = ros_bag_id
        self.language_instruction = language_instruction
        self.duration = duration
        self.morphology = morphology
        self.robot_type = robot_type
        self.effector_type = effector_type
        self.device_number = device_number
        self.resource_type = resource_type
        self.file_path = file_path
        self.status = status
        self.task_id = task_id
        self.editor = editor
        self.description = description
        self.tag = tag
        self.version_num = version_num
        self.create_time = create_time
        self.update_time = update_time

修改前的find_by_id方法 #

def find_by_id(self, id : int) -> Optional[Standard_Data_Model]:
    data = self.session.execute(
        select(Standard_Data_Model).where(Standard_Data_Model.id == id)
    )
    res = data.scalars().first()
    return res

修改后的find_by_id方法 #

def find_by_id(self, id : int, _filter: str) -> Optional[Standard_Data_Model]:
    if _filter:
        data = self.session.execute(
            select(Standard_Data_Model).where(
                (Standard_Data_Model.id == id) & (Standard_Data_Model.language_instruction == _filter)
            )
        )
    else:
        data = self.session.execute(
            select(Standard_Data_Model).where(Standard_Data_Model.id == id)
        )
    res = data.scalars().first()
    return res

问题原因 #

错误本质是FastAPI默认使用Pydantic序列化返回值，但Pydantic无法直接处理SQLAlchemy的ORM实例。之前代码能正常运行，说明之前调用find_by_id的路由逻辑中，存在将ORM实例转换为Pydantic模型或字典的步骤；而修改方法后，调用该方法的路由遗漏了这个转换步骤，直接返回了SQLAlchemy的ORM对象，导致序列化失败。

修改后的find_by_id方法本身逻辑无问题，返回的仍然是Standard_Data_Model实例，问题出在方法的调用端，而非方法本身。

解决办法 #

方法1：创建对应Pydantic模型（推荐） #

为SQLAlchemy模型定义对应的Pydantic模型，利用Pydantic的ORM适配能力完成转换：

from pydantic import BaseModel
from datetime import datetime
from typing import Optional

class StandardDataModelSchema(BaseModel):
    id: Optional[int] = None
    ros_bag_id: Optional[int] = None
    language_instruction: Optional[str] = None
    duration: Optional[str] = None
    morphology: Optional[str] = None
    robot_type: Optional[str] = None
    effector_type: Optional[str] = None
    device_number: Optional[str] = None
    resource_type: Optional[str] = None
    file_path: Optional[str] = None
    status: Optional[str] = None
    task_id: Optional[str] = None
    editor: Optional[str] = None
    description: Optional[str] = None
    tag: Optional[str] = None
    version_num: Optional[str] = None
    create_time: Optional[datetime] = None
    update_time: Optional[datetime] = None

    class Config:
        from_attributes = True  # Pydantic v2版本使用，v1版本替换为orm_mode=True

在路由中使用该Pydantic模型转换返回值：

@app.get("/data/{id}")
def get_data(id: int, _filter: Optional[str] = None):
    dal = StandardDataDAL(session)
    data = dal.find_by_id(id, _filter)
    return StandardDataModelSchema.from_orm(data)

方法2：将ORM实例转换为字典返回 #

直接将SQLAlchemy实例转为字典，跳过Pydantic模型定义（适合简单场景）：

from sqlalchemy import inspect

def orm_to_dict(obj):
    # 仅提取模型的字段属性，排除SQLAlchemy内部属性
    return {c.key: getattr(obj, c.key) for c in inspect(obj).mapper.column_attrs}

# 路由中使用
@app.get("/data/{id}")
def get_data(id: int, _filter: Optional[str] = None):
    dal = StandardDataDAL(session)
    data = dal.find_by_id(id, _filter)
    return orm_to_dict(data) if data else None

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