Marshmallow实现SQL Server JSON字符串与字典双向映射求助

数据库表结构 #

COLUMN_NAME              ORDINAL_POSITION DATA_TYPE       
----------------------------------------------------
id                       1 varchar
collection_id            2 varchar
created_at               3 datetimeoffset  
mimetype                 4 varchar
size                     5 int
file_hash                6 varchar
storageKey               7 varchar
extra_data               8 varchar

当前Marshmallow Schema定义 #

class ImageDisplaySchema(Schema):
    id = fields.Str()
    created_at = fields.Str()
    collection_id = fields.Str()
    mimetype = fields.Str()
    size = fields.Int()
    extra_data = fields.Dict()

遇到的问题 #

尝试通过pre_dump/pre_load实现extra_data列（数据库存储为JSON字符串）与字典字段的双向转换时，出现两种异常：

• 使用json.loads()在pre_dump中转换JSON字符串为字典后，会修改原始模型实例的属性，导致后续写入数据库时尝试存入字典类型，触发报错：

cartridgeocr-annotations-1           |   File "src/pymssql/_mssql.pyx", line 1976, in pymssql._mssql._quote_or_flatten
cartridgeocr-annotations-1           | ValueError: expected a simple type, a tuple or a list

• 若改为json.dumps()，在调用dump()返回查询结果时，会因尝试将JSON字符串反序列化为字典失败，触发报错：

cartridgeocr-annotations-1           | ValueError: dictionary update sequence element #0 has length 1; 2 is required

解决方案 #

方法一：自定义字段（推荐） #

创建自定义字段处理JSON字符串与字典的双向转换，不修改原始模型实例属性，更符合Marshmallow设计逻辑：

import json
from marshmallow import fields, Schema

class JSONStringDictField(fields.Field):
    def _serialize(self, value, attr, obj, **kwargs):
        # 序列化：数据库的JSON字符串 -> 前端所需字典
        if value is None:
            return {}
        return json.loads(value)
    
    def _deserialize(self, value, attr, data, **kwargs):
        # 反序列化：前端传入的字典 -> 数据库存储的JSON字符串
        if value is None:
            return json.dumps({})
        return json.dumps(value)

# 更新Schema使用自定义字段
class ImageDisplaySchema(Schema):
    id = fields.Str()
    created_at = fields.Str()
    collection_id = fields.Str()
    mimetype = fields.Str()
    size = fields.Int()
    extra_data = JSONStringDictField()

方法二：修正pre_dump/pre_load实现 #

若坚持使用钩子函数，需避免修改原始模型实例，而是返回处理后的字典副本：

import json
from marshmallow import fields, Schema, pre_dump, pre_load

class ImageDisplaySchema(Schema):
    id = fields.Str()
    created_at = fields.Str()
    collection_id = fields.Str()
    mimetype = fields.Str()
    size = fields.Int()
    extra_data = fields.Dict()

    @pre_dump
    def serialize_extra_data(self, data, many):
        # 转换模型实例为字典并处理extra_data，不修改原始实例
        data_dict = {key: value for key, value in data.__dict__.items() if not key.startswith('_')}
        if data_dict.get('extra_data'):
            data_dict['extra_data'] = json.loads(data_dict['extra_data'])
        return data_dict

    @pre_load
    def deserialize_extra_data(self, data, many):
        # 反序列化时将字典转为JSON字符串
        if data.get('extra_data'):
            data['extra_data'] = json.dumps(data['extra_data'])
        return data

使用这两种方法后，即可实现extra_data字段在数据库JSON字符串与前端字典之间的无缝双向转换，同时避免触发数据库写入或序列化时的异常。

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