Python嵌套类序列化后无法存入MongoDB(PyMongo)的问题咨询
嵌套Python类存入MongoDB(PyMongo)的问题解决
问题背景
尝试将Python嵌套类数据序列化后存入MongoDB(使用PyMongo),已在Categories类实现toJSON()方法,输出JSON格式正常,但调用insert_one()无法保存数据。
相关代码
Categories类
class Categories: def __init__(self, store_key): self.store_key = store_key self.categories = list() # 元素为Category()对象 def toJSON(self): return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)
Category类
class Category: def __init__(self): self.name = '' self.key = '' self.subcategories = []
问题原因
- 直接传
Categories实例给insert_one():PyMongo无法识别自定义类对象,它只支持BSON可序列化的原生Python类型(字典、列表、基础数据类型等)。 - 传
toJSON()返回的字符串:MongoDB存储的是BSON结构,而非纯JSON字符串,直接插入字符串会被当成普通文本存储,且不符合你想要的嵌套结构需求。
可行性说明
完全可以实现嵌套结构的完整存储,只要把自定义类转换成PyMongo能识别的原生Python数据结构(字典嵌套列表/字典)即可,不会丢失任何结构信息。
解决方法
方法1:转成Python字典而非JSON字符串(推荐)
修改序列化逻辑,返回嵌套字典结构,让PyMongo直接处理:
class Categories: def __init__(self, store_key): self.store_key = store_key self.categories = list() def to_dict(self): # 递归处理所有嵌套的自定义类 def obj_convert(obj): if isinstance(obj, (Categories, Category)): res = obj.__dict__.copy() # 遍历处理列表类型的嵌套属性 for k, v in res.items(): if isinstance(v, list): res[k] = [obj_convert(item) for item in v] return res return obj return obj_convert(self) class Category: def __init__(self): self.name = '' self.key = '' self.subcategories = []
使用示例:
# 构造测试数据 phone_cat = Category() phone_cat.name = "手机" phone_cat.key = "phone" electronics_cat = Category() electronics_cat.name = "电子产品" electronics_cat.key = "electronics" electronics_cat.subcategories.append(phone_cat) store_cats = Categories("store_001") store_cats.categories.append(electronics_cat) # 插入MongoDB from pymongo import MongoClient client = MongoClient() db = client.test_db db.categories.insert_one(store_cats.to_dict())
方法2:将JSON字符串转回字典再插入
如果不想修改原有的toJSON()方法,可以先把JSON字符串反序列化为字典:
json_str = store_cats.toJSON() data_dict = json.loads(json_str) db.categories.insert_one(data_dict)
这种方法多了一次序列化+反序列化的过程,效率略低于方法1。
方法3:自定义BSON编码器
如果需要频繁处理这类自定义类,可以让PyMongo自动识别并转换:
from bson import json_util from pymongo import MongoClient def custom_encoder(obj): if isinstance(obj, (Categories, Category)): res = obj.__dict__.copy() for k, v in res.items(): if isinstance(v, list): res[k] = [custom_encoder(item) for item in v] return res raise TypeError(f"无法序列化类型: {type(obj)}") # 注册自定义编码器 json_util.DEFAULT_JSON_OPTIONS = json_util.JSON_OPTIONS.with_default_encoder(custom_encoder) # 直接插入自定义类实例 client = MongoClient() db = client.test_db db.categories.insert_one(store_cats)
内容的提问来源于stack exchange,提问作者AnkurS
相关产品推荐
相关产品推荐

