from __future__ import annotations
import io
import typing
from typing import TypeVar
from babel.support import LazyProxy
from .fields import BaseField
from ..utils import json
from ..utils.mixins import ContextInstanceMixin
if typing.TYPE_CHECKING:
from ..bot.bot import Bot
__all__ = ('MetaTelegramObject', 'TelegramObject', 'InputFile', 'String', 'Integer', 'Float', 'Boolean')
PROPS_ATTR_NAME = '_props'
VALUES_ATTR_NAME = '_values'
ALIASES_ATTR_NAME = '_aliases'
# Binding of builtin types
InputFile = TypeVar('InputFile', 'InputFile', io.BytesIO, io.FileIO, str)
String = TypeVar('String', bound=str)
Integer = TypeVar('Integer', bound=int)
Float = TypeVar('Float', bound=float)
Boolean = TypeVar('Boolean', bound=bool)
T = TypeVar('T')
[docs]class TelegramObject(ContextInstanceMixin, metaclass=MetaTelegramObject):
"""
Abstract class for telegram objects
"""
def __init__(self, conf: typing.Dict[str, typing.Any]=None, **kwargs: typing.Any) -> None:
"""
Deserialize object
:param conf:
:param kwargs:
"""
if conf is None:
conf = {}
self._conf = conf
# Load data
for key, value in kwargs.items():
if key in self.props:
self.props[key].set_value(self, value, parent=self)
else:
self.values[key] = value
# Load default values
for key, value in self.props.items():
if value.default and key not in self.values:
self.values[key] = value.default
@property
def conf(self) -> typing.Dict[str, typing.Any]:
return self._conf
@property
def props(self) -> typing.Dict[str, BaseField]:
"""
Get props
:return: dict with props
"""
return getattr(self, PROPS_ATTR_NAME, {})
@property
def props_aliases(self) -> typing.Dict[str, str]:
"""
Get aliases for props
:return:
"""
return getattr(self, ALIASES_ATTR_NAME, {})
@property
def values(self) -> typing.Dict[str, typing.Any]:
"""
Get values
:return:
"""
if not hasattr(self, VALUES_ATTR_NAME):
setattr(self, VALUES_ATTR_NAME, {})
return getattr(self, VALUES_ATTR_NAME)
@property
def telegram_types(self) -> typing.List[TelegramObject]:
return type(self).telegram_types
[docs] @classmethod
def to_object(cls: typing.Type[T], data: typing.Dict[str, typing.Any]) -> T:
"""
Deserialize object
:param data:
:return:
"""
return cls(**data)
@property
def bot(self) -> Bot:
from ..bot.bot import Bot
bot = Bot.get_current()
if bot is None:
raise RuntimeError("Can't get bot instance from context. "
"You can fix it with setting current instance: "
"'Bot.set_current(bot_instance)'")
return bot
[docs] def to_python(self) -> typing.Dict[str, typing.Any]:
"""
Get object as JSON serializable
:return:
"""
result = {}
for name, value in self.values.items():
if value is None:
continue
if name in self.props:
value = self.props[name].export(self)
if isinstance(value, TelegramObject):
value = value.to_python()
if isinstance(value, LazyProxy):
value = str(value)
result[self.props_aliases.get(name, name)] = value
return result
[docs] def clean(self) -> None:
"""
Remove empty values
"""
for key, value in self.values.copy().items():
if value is None:
del self.values[key]
[docs] def as_json(self) -> str:
"""
Get object as JSON string
:return: JSON
:rtype: :obj:`str`
"""
return json.dumps(self.to_python())
@classmethod
def create(cls: typing.Type[T], *args: typing.Any, **kwargs: typing.Any) -> T:
raise NotImplemented
def __str__(self) -> str:
"""
Return object as string. Alias for '.as_json()'
:return: str
"""
return self.as_json()
def __getitem__(self, item: typing.Union[str, int]) -> typing.Any:
"""
Item getter (by key)
:param item:
:return:
"""
if item in self.props:
return self.props[item].get_value(self)
raise KeyError(item)
def __setitem__(self, key: str, value: typing.Any) -> None:
"""
Item setter (by key)
:param key:
:param value:
:return:
"""
if key in self.props:
return self.props[key].set_value(self, value, self.conf.get('parent', None))
raise KeyError(key)
def __contains__(self, item: str) -> bool:
"""
Check key contains in that object
:param item:
:return:
"""
# self.clean()
return bool(self.values.get(item, None))
def __iter__(self) -> typing.Iterator[str]:
"""
Iterate over items
:return:
"""
for item in self.to_python().items():
yield item
[docs] def iter_keys(self) -> typing.Generator[typing.Any, None, None]:
"""
Iterate over keys
:return:
"""
for key, _ in self:
yield key
[docs] def iter_values(self) -> typing.Generator[typing.Any, None, None]:
"""
Iterate over values
:return:
"""
for _, value in self:
yield value
def __hash__(self) -> int:
def _hash(obj) -> int:
buf: int = 0
if isinstance(obj, list):
for item in obj:
buf += _hash(item)
elif isinstance(obj, dict):
for dict_key, dict_value in obj.items():
buf += hash(dict_key) + _hash(dict_value)
else:
try:
buf += hash(obj)
except TypeError: # Skip unhashable objects
pass
return buf
result = 0
for key, value in sorted(self.values.items()):
result += hash(key) + _hash(value)
return result
def __eq__(self, other: TelegramObject) -> bool:
return isinstance(other, self.__class__) and hash(other) == hash(self)