Source code for telegram_bot_discussion.ext.coder.store_coder

from abc import ABC, abstractmethod
from hashlib import md5
from typing import Any, List, Tuple


from ...button.coder_interface import CoderInterface


[docs] class KeyValueStorageInterface(ABC): """KeyValueStorageInterface is interface for external storage where custom values of `Params` are stored."""
[docs] @staticmethod @abstractmethod def autogenerate_primary_key(): ...
[docs] @abstractmethod def get(self, key: str) -> Any: ...
[docs] @abstractmethod def add(self, value: Any) -> str: ...
[docs] class StoreCoder(CoderInterface): """`StoreCoder` StoreCoder serializes data and saves its value in an external storage with unique primary key. The primary key is stored in the `Button`. For deserialization, the data is got from the external storage by primary key. So there are no any limits for callback query data (long strings, complex multi params, redundant big json and etc.). """ db: KeyValueStorageInterface """ KeyValueStorageInterface: external storage where custom values of `Params` are stored.""" secret: str """ str: Secret string is used for making digital signature of data.""" def __init__(self, secret: str, db: KeyValueStorageInterface): self.secret = secret self.db = db
[docs] def serialize(self, *args: Any) -> str: return self.db.add(args)
[docs] def deserialize(self, data: str) -> List[Any]: return self.db.get(data)
[docs] def get_signature(self, chat_id: int, serialized_data: str, sender_id: int) -> str: return md5( f"{serialized_data}{self.secret}{chat_id}{sender_id}".encode() ).hexdigest()
[docs] def sign(self, signature: str, serialized_data: str) -> str: return f"{signature}{serialized_data}"
[docs] def extract_signature_and_serialized_data( self, signed_serialized_data: str ) -> Tuple[str, str]: signature = signed_serialized_data[:32] serialized_data = signed_serialized_data[32:] return signature, serialized_data
[docs] def extract_signature_and_deserialized_data( self, signed_serialized_data: str ) -> Tuple[str, List[Any]]: signature, serialized_data = self.extract_signature_and_serialized_data( signed_serialized_data ) deserialized_data = self.deserialize(serialized_data) return signature, deserialized_data