Question 335755 in Telegram API Question posted in
A comprehensive official documentation can be found here.

Telegram – Is it possible save peewee SQL query as string or json and then load back to ModelSelect object

I created a universal keyboard with pagination in telegram bot. It works on a peewee query object ModelSelect, but I can’t save it to redis or other storage, now I save it in memory, and every run the object is erased and the keyboard stops working.

I tried:

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

# create User table
User.create_table()

# add new User
User(name = "new_user").save()

# select User with id = 1
item = User.select().where(User.id == 1)
print(f"SQL: {item.sql()}")

# try to convert ModelSelect to dict
_dict = model_to_dict(item)
print(_dict)

# try to convert dict to ModelSelect
_item = dict_to_model(User, _dict)
print(_item)

But got error:

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:Python310libsite-packagesplayhouseshortcuts.py", line 74, in model_to_dict
    for field in model._meta.sorted_fields:
AttributeError: 'ModelSelect' object has no attribute '_meta'
Tags:

2

Answers


  1. Chosen as BEST ANSWER
    0

    I solved my problem. But it may cause error in complex query with joins, but for small tests it looks fine.

    from peewee import ModelSelect, Expression
from peewee import * 
from typing import *

# create database
db = SqliteDatabase(':memory:')

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

# create User table
User.create_table()

# add new User
User(name = "new_user").save()

class ModelSelectJSONify:

    @staticmethod
    # function for export ModelSelect to dict
    def dump(item: ModelSelect) -> dict:
        result = {}
        for key, value in item.__dict__.items():
            if key in ('_database', '_from_list', '_returning'):
                continue

            if not isinstance(value, (str, int, float, bool, dict)) and value is not None:
                
                if isinstance(value, Expression):
                    value = dict(value_type = "Expression",
                                 value = ModelSelectJSONify.dump(value))
                elif isinstance(value, list):
                    value = list(map(ModelSelectJSONify._dump_name, value))
                else:
                    value = ModelSelectJSONify._dump_name(value)

            result[key] = value
        
        return result


    @staticmethod
    # function for import dict back to Any
    def _dump_name(value: Any) -> str:
        if hasattr(value, '__name__'):
            return f"MODEL {value.__name__}"
        else:
            return f"MODEL_PROPERTY {value.model.__name__}.{value.name}"

    @staticmethod
    # function for import dict back to Any
    def _load_name(model: Model, value: str) -> Any:
        if value.startswith(f"MODEL_PROPERTY {model.__name__}"):
            return getattr(model, value.replace(f"MODEL_PROPERTY {model.__name__}.", ""))
        
        elif value.startswith(f"MODEL {model.__name__}"):
            return model
        
        else:
            return value
        
    @staticmethod
    # function for import dict back to Any
    def _direct_load(obj: Any, data: dict, model: Model = None) -> Any:
        result = {}
        for key, value in data.items():
            if key in ('model'):
                continue

            if isinstance(value, dict) and value.get("value_type"):
                match value["value_type"]:
                    case "Expression":
                        value = ModelSelectJSONify._direct_load(Expression, value["value"], model = model)
            
            if isinstance(value, str) and model is not None:
                value = ModelSelectJSONify._load_name(model, value)
            elif isinstance(value, list) and model is not None:
                value = [ModelSelectJSONify._load_name(model, x) for x in value]

            result[key] = value

        return obj(**result)

    @staticmethod
    # function for import dict back to ModelSelect
    def load(model: Model, data: dict) -> ModelSelect:
        result = model.select()
        for key, value in data.items():
            if key in ('model'):
                continue

            if isinstance(value, dict) and value.get("value_type"):
                match value["value_type"]:
                    case "Expression":
                        value = ModelSelectJSONify._direct_load(Expression, value["value"], model = model)
            
            if isinstance(value, str) and model is not None:
                value = ModelSelectJSONify._load_name(model, value)

            elif isinstance(value, list) and model is not None:
                value = [ModelSelectJSONify._load_name(model, x) for x in value]

            setattr(result, key, value)
        return result

def modelselect_to_dict(item: ModelSelect) -> dict:
    return ModelSelectJSONify.dump(item)

def dict_to_modelselect(model: Model, data: dict) -> ModelSelect:
    return ModelSelectJSONify.load(model, data)

item = User.select().where((User.id == 1) & (User.name == "new_user")).limit(1).offset(0).group_by(User.id)
dicted = modelselect_to_dict(item)
print("Exported as dict:", dicted)

usered = dict_to_modelselect(User, dicted)

print("Initial SQL:", item.sql())
# ('SELECT "t1"."id", "t1"."name" FROM "user" AS "t1" WHERE (("t1"."id" = ?) AND ("t1"."name" = ?)) GROUP BY "t1"."id" LIMIT ? OFFSET ?', [1, 'new_user', 1, 0])
print("Imported SQL:", usered.sql())
# ('SELECT "t1"."id", "t1"."name" FROM "user" AS "t1" WHERE (("t1"."id" = ?) AND ("t1"."name" = ?)) GROUP BY "t1"."id" LIMIT ? OFFSET ?', [1, 'new_user', 1, 0])

print("Equal:", item.__dict__ == usered.__dict__)
# True

  2. 0

    This is the wrong idea. Serializing model instances to dicts is perfectly fine, and they can be de-serialized back to model instances. You cannot reasonably expect to serialize a query object, though. The query just represents the SQL and, when executed, contains a reference to a cursor object.

    If you want to serialize an object, you would:

    # select User with id = 1 -- note call to .get()!!
item = User.select().where(User.id == 1).get()
data = model_to_dict(item)
    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search