FastAPI从MySQL获取数据时serialize_response错误的解决方法

问题现象 #

1. /get-sensors/接口设置response_model=List[Data]，但返回{"status": "success", "list": data}时触发ResponseValidationError，提示输入应为合法列表。
2. MySQL返回的Decimal类型无法匹配Data模型的float字段，datetime类型无法匹配str字段。

错误原因 #

• 响应模型不匹配：接口声明返回List[Data]类型，但实际返回的是包含status和list键的字典，类型校验失败。
• 类型转换缺失：mysql-connector返回的Decimal和datetime对象，Pydantic默认无法自动转换为float和str类型。

修复方案 #

1. 定义正确的响应模型 #

创建包含状态和数据列表的响应模型，替代原来的List[Data]。

2. 处理类型转换 #

对Data模型添加字段验证器，自动将Decimal转为float，datetime转为指定格式的字符串。

3. 调整接口返回逻辑 #

确保返回的数据结构与新的响应模型一致。

修改后的完整代码 #

import mysql.connector
from fastapi import FastAPI, HTTPException
from typing import Optional, List
from pydantic import BaseModel, field_validator
from datetime import datetime, timedelta
from decimal import Decimal


def get_db_connection():
    return mysql.connector.connect(
        user='*****',
        password='*****',
        host='*****',
        port=*****, 
        database='******'
    )


class Data(BaseModel):
    timestamp: Optional[str] = None
    acceleration_x: Optional[float] = None
    acceleration_y: Optional[float] = None
    acceleration_z: Optional[float] = None
    velocity_x: Optional[float] = None
    velocity_y: Optional[float] = None
    velocity_z: Optional[float] = None
    x: Optional[float] = None
    y: Optional[float] = None
    z: Optional[float] = None
    angular_accelerltion_x: Optional[float] = None
    angular_accelerltion_y: Optional[float] = None
    angular_accelerltion_z: Optional[float] = None
    angular_velocity_x: Optional[float] = None
    angular_velocity_y: Optional[float] = None
    angular_velocity_z: Optional[float] = None
    orientation_x: Optional[float] = None
    orientation_y: Optional[float] = None
    orientation_z: Optional[float] = None
    temperature: Optional[float] = None

    @field_validator('acceleration_x', 'acceleration_y', 'acceleration_z', 
                    'velocity_x', 'velocity_y', 'velocity_z', 'x', 'y', 'z',
                    'angular_accelerltion_x', 'angular_accelerltion_y', 'angular_accelerltion_z',
                    'angular_velocity_x', 'angular_velocity_y', 'angular_velocity_z',
                    'orientation_x', 'orientation_y', 'orientation_z', 'temperature',
                    mode='before')
    def decimal_to_float(cls, v):
        if isinstance(v, Decimal):
            return float(v)
        return v

    @field_validator('timestamp', mode='before')
    def datetime_to_str(cls, v):
        if isinstance(v, datetime):
            return v.strftime("%Y-%m-%d %H:%M:%S")
        return v

# 定义包含状态的响应模型
class SensorResponse(BaseModel):
    status: str
    list: List[Data]


fibo_sub = FastAPI()


@fibo_sub.post("/insert_sensors/")
def insert_sensors(data: Data):
    db = get_db_connection()
    mycurser = db.cursor()
    insert_query = (
        "INSERT INTO sensors(timestamp,acceleration_x,acceleration_y,acceleration_z,velocity_x,velocity_y,velocity_z,x,y,z,angular_accelerltion_x,angular_accelerltion_y,angular_accelerltion_z,angular_velocity_x,angular_velocity_y,angular_velocity_z,orientation_x,orientation_y,orientation_z,temperature) "
        " VALUES(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);"
    )
    vals = (data.timestamp, data.acceleration_x, data.acceleration_y, data.acceleration_z, data.velocity_x, data.velocity_y, data.velocity_z, data.x, data.y, data.z, data.angular_accelerltion_x,
            data.angular_accelerltion_y, data.angular_accelerltion_z, data.angular_velocity_x, data.angular_velocity_y, data.angular_velocity_z, data.orientation_x, data.orientation_y, data.orientation_z, data.temperature)
    mycurser.execute(insert_query, vals)
    db.commit()
    return {"status": "success"}


@fibo_sub.get("/get-sensors/", response_model=SensorResponse)
def get_sensors(start_date: str):
    db = get_db_connection()
    mycurser = db.cursor(dictionary=True)
    start_time = datetime.strptime(start_date, "%Y-%m-%d %H:%M:%S")
    end_time = start_time - timedelta(seconds=60)

    query = "SELECT * FROM sensors WHERE timestamp BETWEEN %s AND %s"
    mycurser.execute(query, (end_time, start_time))

    data = mycurser.fetchall()
    
    return {"status": "success", "list": data}

版本兼容说明 #

如果使用Pydantic v1，需将field_validator替换为validator装饰器，语法调整为：

from pydantic import validator

@validator('acceleration_x', ..., pre=True)
def decimal_to_float(cls, v):
    if isinstance(v, Decimal):
        return float(v)
    return v

内容的提问来源于stack exchange，提问作者Amirmohammad Alian nezhad