Flask SocketIO传递SQLAlchemy模型数据时遇序列化错误求解决

尝试将SQLAlchemy模型序列化后，通过Flask SocketIO传递给HTML客户端，采用先转字典再用jsonify的方法，却触发错误：TypeError: Object of type Response is not JSON serializable。

相关代码（app.py） #

from flask import Flask, render_template, jsonify
from flask_marshmallow import Marshmallow
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import secrets
import string
import logging
from flask_socketio import SocketIO, emit
import os
import subprocess

async_mode = None

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///filename.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

socketio_obj = SocketIO(app)   

class JobQueue(db.Model):
    __tablename__ = 'job_queue'
    job_id = db.Column(db.Integer, primary_key=True)
    unique_job_key = db.Column(db.String(64), index=True)
    user_name = db.Column(db.String, index=True)
    input_string = db.Column(db.String(256))
    is_done = db.Column(db.Boolean)
    created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)

    def to_dictionary(self):
        return {
            "job_id": self.job_id,
            "unique_job_key": self.unique_job_key,
            "user_name": self.user_name,
            "input_string": self.input_string,
            "is_done": self.is_done,
            "created_at": self.created_at
        }

def get_data():
    users = JobQueue.query.order_by(JobQueue.created_at).all()
    return users

def get_serialized_data():
    all_users = get_data()
    returns = [item.to_dictionary() for item in all_users]
    return jsonify(all_users = returns)

@socketio_obj.on('first_connect_event', namespace='/test')
def handle_first_connect_event(arg1):
    try:
        msg = arg1['first_message']
        print(msg)
        users = get_serialized_data()
        emit('get_data', {'users': users}) # 报错行
    except Exception as ex:
        logging.exception("")
        print(ex)

完整错误栈 #

ERROR:root:
Traceback (most recent call last):
  File "C:\git\funkclusterfrontend\app.py", line 106, in handle_first_connect_event
    emit('get_data', {'users': users})
  File "C:\ProgramData\Miniconda3\lib\site-packages\flask_socketio\__init__.py", line 899, in emit
    return socketio.emit(event, *args, namespace=namespace, to=to,
  File "C:\ProgramData\Miniconda3\lib\site-packages\flask_socketio\__init__.py", line 460, in emit
    self.server.emit(event, *args, namespace=namespace, to=to,
  File "C:\ProgramData\Miniconda3\lib\site-packages\socketio\server.py", line 294, in emit
    self.manager.emit(event, data, namespace, room=room,
  File "C:\ProgramData\Miniconda3\lib\site-packages\socketio\base_manager.py", line 167, in emit
    self.server._emit_internal(eio_sid, event, data, namespace, id)
  File "C:\ProgramData\Miniconda3\lib\site-packages\socketio\server.py", line 612, in _emit_internal
    self._send_packet(eio_sid, packet.Packet(
  File "C:\ProgramData\Miniconda3\lib\site-packages\socketio\server.py", line 617, in _send_packet
    encoded_packet = pkt.encode()
  File "C:\ProgramData\Miniconda3\lib\site-packages\socketio\packet.py", line 61, in encode
    encoded_packet += self.json.dumps(data, separators=(',', ':'))
  File "C:\ProgramData\Miniconda3\lib\json\__init__.py", line 234, in dumps
    return cls(
  File "C:\ProgramData\Miniconda3\lib\json\encoder.py", line 199, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "C:\ProgramData\Miniconda3\lib\json\encoder.py", line 257, in iterencode
    return _iterencode(o, 0)
  File "C:\ProgramData\Miniconda3\lib\json\encoder.py", line 179, in default
    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type Response is not JSON serializable
Object of type Response is not JSON serializable

核心问题分析 #

jsonify()是Flask用于生成HTTP响应的函数，返回的是Response对象而非纯JSON数据。SocketIO的emit()方法会自动对传入的Python数据结构做JSON序列化，不需要提前用jsonify包装，否则会尝试序列化Response对象，导致报错。另外，datetime对象默认无法被JSON序列化，需要额外处理。

修复方案 #

1. 移除get_serialized_data()中的jsonify调用，直接返回字典列表
2. 将created_at字段从datetime对象转换为可序列化的字符串格式（如ISO格式）

修改后的代码 #

方案1：修改to_dictionary方法直接处理日期

class JobQueue(db.Model):
    # ... 其他字段省略 ...
    def to_dictionary(self):
        return {
            "job_id": self.job_id,
            "unique_job_key": self.unique_job_key,
            "user_name": self.user_name,
            "input_string": self.input_string,
            "is_done": self.is_done,
            "created_at": self.created_at.isoformat()  # 转换为ISO格式字符串
        }

def get_serialized_data():
    all_users = get_data()
    return [item.to_dictionary() for item in all_users]  # 直接返回字典列表

方案2：在get_serialized_data中处理日期

def get_serialized_data():
    all_users = get_data()
    serialized_list = []
    for item in all_users:
        user_dict = item.to_dictionary()
        user_dict["created_at"] = user_dict["created_at"].isoformat()
        serialized_list.append(user_dict)
    return serialized_list

最后修改SocketIO事件处理函数

@socketio_obj.on('first_connect_event', namespace='/test')
def handle_first_connect_event(arg1):
    try:
        msg = arg1['first_message']
        print(msg)
        users = get_serialized_data()
        emit('get_data', {'users': users})  # 直接传入序列化后的列表
    except Exception as ex:
        logging.exception("")
        print(ex)

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