在Peewee ORM的使用场景中，我们常常需要将查询对象进行序列化存储或者网络传输，之后再通过反序列化还原为可执行的查询对象，避免重复构建查询逻辑的繁琐。Peewee的查询对象本身包含模型信息、筛选条件、排序规则等结构化数据，直接序列化会丢失部分内部状态，因此需要采用特定的处理方式。
序列化前的准备
首先需要明确，Peewee的查询对象（ModelSelect类型）不能直接通过json.dumps序列化，因为其包含不可序列化的模型类引用、数据库连接对象等。我们需要先提取查询的核心结构化信息，再进行序列化操作。以下示例基于Peewee的基础模型定义：
import peewee
from peewee import *
# 初始化数据库连接
db = SqliteDatabase('test.db')
# 定义示例模型
class User(Model):
name = CharField()
age = IntegerField()
email = CharField()
class Meta:
database = db
# 创建表（实际使用时根据需求调整）
db.connect()
db.create_tables([User], safe=True)
将Peewee查询对象序列化为字符串或JSON
序列化的核心思路是提取查询的模型名称、筛选条件、排序规则、分页参数等信息，将这些信息转为可序列化的字典，再转为JSON字符串。我们可以通过自定义方法来解析查询对象的结构：
import json
def serialize_query(query):
"""将Peewee查询对象序列化为可JSON化的字典"""
# 获取查询对应的模型名称
model = query.model
model_name = model.__name__
# 提取筛选条件，这里简化提取为字符串形式，实际可根据需求解析为结构化数据
# 注意：复杂条件可能需要更复杂的解析逻辑
where_clauses = []
for clause in query._where:
# 简化处理，仅提取字段名和对应值，实际场景需适配更多条件类型
if isinstance(clause, tuple):
field_name = clause[0].name
op = str(clause[1])
value = clause[2]
where_clauses.append({
'field': field_name,
'op': op,
'value': value
})
# 提取排序规则
order_by = []
for order in query._order_by:
field_name = order[0].name
direction = 'ASC' if order[1] else 'DESC'
order_by.append({'field': field_name, 'direction': direction})
# 提取分页参数
limit = query._limit
offset = query._offset
return {
'model': model_name,
'where': where_clauses,
'order_by': order_by,
'limit': limit,
'offset': offset
}
# 示例：构建查询并序列化
query = User.select().where(User.age > 18).order_by(User.name.asc()).limit(10)
query_dict = serialize_query(query)
# 转为JSON字符串
query_json = json.dumps(query_dict, ensure_ascii=False)
print("序列化后的JSON字符串：", query_json)
将序列化数据反序列化为Peewee查询对象
反序列化的过程是根据之前提取的结构化信息，重新构建对应的Peewee查询对象。我们需要先根据模型名称找到对应的模型类，再逐步还原筛选条件、排序规则等：
def deserialize_query(query_dict):
"""将序列化后的字典还原为Peewee查询对象"""
# 根据模型名称获取模型类
model_name = query_dict['model']
# 假设当前命名空间中存在对应的模型类，实际可根据项目结构调整为动态导入
model = globals()[model_name]
# 初始化查询对象
query = model.select()
# 还原筛选条件
for where_item in query_dict['where']:
field = getattr(model, where_item['field'])
op = where_item['op']
value = where_item['value']
# 根据操作符构建条件，这里仅示例等于和大于的情况，其他操作符可扩展
if '>' in op:
query = query.where(field > value)
elif '==' in op:
query = query.where(field == value)
# 更多操作符可根据实际需求补充
# 还原排序规则
for order_item in query_dict['order_by']:
field = getattr(model, order_item['field'])
if order_item['direction'] == 'ASC':
query = query.order_by(field.asc())
else:
query = query.order_by(field.desc())
# 还原分页参数
if query_dict['limit'] is not None:
query = query.limit(query_dict['limit'])
if query_dict['offset'] is not None:
query = query.offset(query_dict['offset'])
return query
# 示例：反序列化JSON字符串为查询对象
restored_query = deserialize_query(json.loads(query_json))
# 执行还原后的查询
for user in restored_query:
print(f"用户姓名：{user.name}，年龄：{user.age}")
注意事项
- 上述示例中的条件解析逻辑较为简化，实际项目中如果查询包含复杂的条件组合（如
|或&逻辑运算、关联查询条件等），需要扩展解析逻辑，确保完整还原所有查询规则。 - 序列化时如果查询中包含动态生成的值（如当前时间、函数调用结果），反序列化时需要保证这些值的可还原性，避免还原后查询结果不一致。
- 如果项目中有多个模型，反序列化时获取模型类的逻辑需要适配项目的模型管理方式，避免硬编码模型映射关系。
总结
Peewee查询对象的序列化与反序列化核心在于提取查询的结构化元信息，避免直接序列化不可序列化的内部对象。通过自定义序列化和反序列化方法，我们可以灵活地将查询对象转为JSON字符串存储或传输，之后完整还原为可执行的查询对象，大幅提升开发效率。实际使用时可根据项目的查询复杂度调整解析逻辑，适配更多场景需求。