能否将Peewee的ModelSelect查询对象序列化后存储再还原？

我在Telegram机器人中实现了一个带分页功能的通用键盘，该键盘依赖Peewee的ModelSelect查询对象，但无法将其保存到Redis或其他存储介质中，目前只能存在内存里，每次重启后对象丢失，键盘功能失效。

我尝试用model_to_dict和dict_to_model转换ModelSelect，代码如下：

from playhouse.shortcuts import model_to_dict, dict_to_model
from peewee import * 

db = SqliteDatabase('test.db')

class User(Model): 
    id: int 
    name: str = TextField()
    
    class Meta:
        database = db

# 创建User表
User.create_table()

# 添加新用户
User(name="new_user").save()

# 查询id=1的用户
item = User.select().where(User.id == 1)
print(f"SQL: {item.sql()}")

# 尝试将ModelSelect转为字典
_dict = model_to_dict(item)
print(_dict)

# 尝试将字典转回ModelSelect
_item = dict_to_model(User, _dict)
print(_item)

但运行后出现错误：

SQL: ('SELECT "t1"."id", "t1"."name" FROM "user" AS "t1" WHERE ("t1"."id" = ?)', [1])
Traceback (most recent call last):
  File "C:\testfile.py", line 27, in <module>
    _dict = model_to_dict(item)
  File "C:\Python310\lib\site-packages\playhouse\shortcuts.py", line 74, in model_to_dict
    for field in model._meta.sorted_fields:
AttributeError: 'ModelSelect' object has no attribute '_meta'

model_to_dict和dict_to_model是用来序列化/反序列化Model实例的工具，而ModelSelect是查询构造器对象，不属于Model实例范畴，因此直接调用会报错。要解决持久化问题，不能直接存储ModelSelect对象本身，而是要序列化查询的核心参数，之后再通过参数重建查询。

通过提取ModelSelect的关键查询参数（模型类、过滤条件、排序规则、分页参数），将其序列化为JSON格式存储，需要时再根据这些参数重新构建查询对象。

示例代码 #

import json
from peewee import *

db = SqliteDatabase('test.db')

class User(Model):
    id = IntegerField(primary_key=True)
    name = TextField()
    
    class Meta:
        database = db

# 序列化查询参数为可存储的字典
def serialize_query(query):
    return {
        'model': query.model.__name__,
        'where': str(query.where_clause) if query.where_clause else None,
        'order_by': [str(expr) for expr in query.order_by] if query.order_by else None,
        'limit': query._limit,
        'offset': query._offset
    }

# 从字典反序列化重建查询
def deserialize_query(data):
    # 维护模型类映射，确保能找到对应的Model
    model_map = {
        'User': User
    }
    model = model_map[data['model']]
    query = model.select()
    
    # 恢复过滤条件
    if data['where']:
        # 注意：eval仅在内部可控场景使用，避免恶意注入风险
        query = query.where(eval(data['where']))
    
    # 恢复排序规则
    if data['order_by']:
        order_exprs = [eval(expr) for expr in data['order_by']]
        query = query.order_by(*order_exprs)
    
    # 恢复分页参数
    if data['limit'] is not None:
        query = query.limit(data['limit'])
    if data['offset'] is not None:
        query = query.offset(data['offset'])
    
    return query

# 测试使用
if __name__ == '__main__':
    User.create_table(safe=True)
    User.create(name="new_user")
    
    # 构造原始查询
    original_query = User.select().where(User.id == 1).limit(10).offset(0)
    print(f"原始查询SQL: {original_query.sql()}")
    
    # 序列化
    serialized_data = serialize_query(original_query)
    print(f"序列化结果: {json.dumps(serialized_data)}")
    
    # 模拟存入Redis
    # import redis
    # r = redis.Redis(host='localhost', port=6379, db=0)
    # r.set('telegram_pagination_query', json.dumps(serialized_data))
    
    # 模拟从Redis读取
    # stored_data = json.loads(r.get('telegram_pagination_query'))
    
    # 反序列化重建查询
    restored_query = deserialize_query(serialized_data)
    print(f"重建后查询SQL: {restored_query.sql()}")

注意事项 #

1. 安全风险：示例中使用eval重建查询条件，仅适合内部可控场景。如果涉及用户输入，建议将条件拆分为字段、操作符、值的结构化数据（如{"field": "id", "op": "=", "value": 1}），再通过Peewee的API动态构建条件，避免注入风险。
2. 复杂查询扩展：如果查询包含JOIN、聚合函数等复杂逻辑，需要扩展序列化逻辑，提取更多查询参数（如关联模型、聚合表达式等）。
3. 模型映射维护：如果有多个Model类，需要在model_map中添加对应的映射关系。

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