mirror of
https://github.com/python-telegram-bot/python-telegram-bot.git
synced 2024-10-23 17:36:26 +02:00
Move Dunder Methods to the Top of Class Bodies (#3883)
Co-authored-by: Hinrich Mahler <22366557+Bibo-Joshi@users.noreply.github.com>
This commit is contained in:
Harshil
GitHub
parent
9c7298c17a
commit
5b0f1697f1
10 changed files with 501 additions and 501 deletions
270
telegram/_bot.py
270
telegram/_bot.py
|
|
@ -309,6 +309,54 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
|
|||
|
||||
self._freeze()
|
||||
|
||||
async def __aenter__(self: BT) -> BT:
|
||||
try:
|
||||
await self.initialize()
|
||||
return self
|
||||
except Exception as exc:
|
||||
await self.shutdown()
|
||||
raise exc
|
||||
|
||||
async def __aexit__(
|
||||
self,
|
||||
exc_type: Optional[Type[BaseException]],
|
||||
exc_val: Optional[BaseException],
|
||||
exc_tb: Optional[TracebackType],
|
||||
) -> None:
|
||||
# Make sure not to return `True` so that exceptions are not suppressed
|
||||
# https://docs.python.org/3/reference/datamodel.html?#object.__aexit__
|
||||
await self.shutdown()
|
||||
|
||||
def __reduce__(self) -> NoReturn:
|
||||
"""Customizes how :func:`copy.deepcopy` processes objects of this type. Bots can not
|
||||
be pickled and this method will always raise an exception.
|
||||
|
||||
.. versionadded:: 20.0
|
||||
|
||||
Raises:
|
||||
:exc:`pickle.PicklingError`
|
||||
"""
|
||||
raise pickle.PicklingError("Bot objects cannot be pickled!")
|
||||
|
||||
def __deepcopy__(self, memodict: Dict[int, object]) -> NoReturn:
|
||||
"""Customizes how :func:`copy.deepcopy` processes objects of this type. Bots can not
|
||||
be deepcopied and this method will always raise an exception.
|
||||
|
||||
.. versionadded:: 20.0
|
||||
|
||||
Raises:
|
||||
:exc:`TypeError`
|
||||
"""
|
||||
raise TypeError("Bot objects cannot be deepcopied!")
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if isinstance(other, self.__class__):
|
||||
return self.bot == other.bot
|
||||
return False
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash((self.__class__, self.bot))
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""Give a string representation of the bot in the form ``Bot[token=...]``.
|
||||
|
||||
|
|
@ -365,6 +413,93 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
|
|||
"""
|
||||
return self._private_key
|
||||
|
||||
@property
|
||||
def request(self) -> BaseRequest:
|
||||
"""The :class:`~telegram.request.BaseRequest` object used by this bot.
|
||||
|
||||
Warning:
|
||||
Requests to the Bot API are made by the various methods of this class. This attribute
|
||||
should *not* be used manually.
|
||||
"""
|
||||
return self._request[1]
|
||||
|
||||
@property
|
||||
def bot(self) -> User:
|
||||
""":class:`telegram.User`: User instance for the bot as returned by :meth:`get_me`.
|
||||
|
||||
Warning:
|
||||
This value is the cached return value of :meth:`get_me`. If the bots profile is
|
||||
changed during runtime, this value won't reflect the changes until :meth:`get_me` is
|
||||
called again.
|
||||
|
||||
.. seealso:: :meth:`initialize`
|
||||
"""
|
||||
if self._bot_user is None:
|
||||
raise RuntimeError(
|
||||
f"{self.__class__.__name__} is not properly initialized. Call "
|
||||
f"`{self.__class__.__name__}.initialize` before accessing this property."
|
||||
)
|
||||
return self._bot_user
|
||||
|
||||
@property
|
||||
def id(self) -> int:
|
||||
""":obj:`int`: Unique identifier for this bot. Shortcut for the corresponding attribute of
|
||||
:attr:`bot`.
|
||||
"""
|
||||
return self.bot.id
|
||||
|
||||
@property
|
||||
def first_name(self) -> str:
|
||||
""":obj:`str`: Bot's first name. Shortcut for the corresponding attribute of
|
||||
:attr:`bot`.
|
||||
"""
|
||||
return self.bot.first_name
|
||||
|
||||
@property
|
||||
def last_name(self) -> str:
|
||||
""":obj:`str`: Optional. Bot's last name. Shortcut for the corresponding attribute of
|
||||
:attr:`bot`.
|
||||
"""
|
||||
return self.bot.last_name # type: ignore
|
||||
|
||||
@property
|
||||
def username(self) -> str:
|
||||
""":obj:`str`: Bot's username. Shortcut for the corresponding attribute of
|
||||
:attr:`bot`.
|
||||
"""
|
||||
return self.bot.username # type: ignore
|
||||
|
||||
@property
|
||||
def link(self) -> str:
|
||||
""":obj:`str`: Convenience property. Returns the t.me link of the bot."""
|
||||
return f"https://t.me/{self.username}"
|
||||
|
||||
@property
|
||||
def can_join_groups(self) -> bool:
|
||||
""":obj:`bool`: Bot's :attr:`telegram.User.can_join_groups` attribute. Shortcut for the
|
||||
corresponding attribute of :attr:`bot`.
|
||||
"""
|
||||
return self.bot.can_join_groups # type: ignore
|
||||
|
||||
@property
|
||||
def can_read_all_group_messages(self) -> bool:
|
||||
""":obj:`bool`: Bot's :attr:`telegram.User.can_read_all_group_messages` attribute.
|
||||
Shortcut for the corresponding attribute of :attr:`bot`.
|
||||
"""
|
||||
return self.bot.can_read_all_group_messages # type: ignore
|
||||
|
||||
@property
|
||||
def supports_inline_queries(self) -> bool:
|
||||
""":obj:`bool`: Bot's :attr:`telegram.User.supports_inline_queries` attribute.
|
||||
Shortcut for the corresponding attribute of :attr:`bot`.
|
||||
"""
|
||||
return self.bot.supports_inline_queries # type: ignore
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
""":obj:`str`: Bot's @username. Shortcut for the corresponding attribute of :attr:`bot`."""
|
||||
return f"@{self.username}"
|
||||
|
||||
@classmethod
|
||||
def _warn(
|
||||
cls, message: str, category: Type[Warning] = PTBUserWarning, stacklevel: int = 0
|
||||
|
|
@ -374,28 +509,6 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
|
|||
"""
|
||||
warn(message=message, category=category, stacklevel=stacklevel + 1)
|
||||
|
||||
def __reduce__(self) -> NoReturn:
|
||||
"""Customizes how :func:`copy.deepcopy` processes objects of this type. Bots can not
|
||||
be pickled and this method will always raise an exception.
|
||||
|
||||
.. versionadded:: 20.0
|
||||
|
||||
Raises:
|
||||
:exc:`pickle.PicklingError`
|
||||
"""
|
||||
raise pickle.PicklingError("Bot objects cannot be pickled!")
|
||||
|
||||
def __deepcopy__(self, memodict: Dict[int, object]) -> NoReturn:
|
||||
"""Customizes how :func:`copy.deepcopy` processes objects of this type. Bots can not
|
||||
be deepcopied and this method will always raise an exception.
|
||||
|
||||
.. versionadded:: 20.0
|
||||
|
||||
Raises:
|
||||
:exc:`TypeError`
|
||||
"""
|
||||
raise TypeError("Bot objects cannot be deepcopied!")
|
||||
|
||||
# TODO: After https://youtrack.jetbrains.com/issue/PY-50952 is fixed, we can revisit this and
|
||||
# consider adding Paramspec from typing_extensions to properly fix this. Currently a workaround
|
||||
def _log(func: Any): # type: ignore[no-untyped-def] # skipcq: PY-D0003
|
||||
|
|
@ -633,111 +746,6 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
|
|||
await asyncio.gather(self._request[0].shutdown(), self._request[1].shutdown())
|
||||
self._initialized = False
|
||||
|
||||
async def __aenter__(self: BT) -> BT:
|
||||
try:
|
||||
await self.initialize()
|
||||
return self
|
||||
except Exception as exc:
|
||||
await self.shutdown()
|
||||
raise exc
|
||||
|
||||
async def __aexit__(
|
||||
self,
|
||||
exc_type: Optional[Type[BaseException]],
|
||||
exc_val: Optional[BaseException],
|
||||
exc_tb: Optional[TracebackType],
|
||||
) -> None:
|
||||
# Make sure not to return `True` so that exceptions are not suppressed
|
||||
# https://docs.python.org/3/reference/datamodel.html?#object.__aexit__
|
||||
await self.shutdown()
|
||||
|
||||
@property
|
||||
def request(self) -> BaseRequest:
|
||||
"""The :class:`~telegram.request.BaseRequest` object used by this bot.
|
||||
|
||||
Warning:
|
||||
Requests to the Bot API are made by the various methods of this class. This attribute
|
||||
should *not* be used manually.
|
||||
"""
|
||||
return self._request[1]
|
||||
|
||||
@property
|
||||
def bot(self) -> User:
|
||||
""":class:`telegram.User`: User instance for the bot as returned by :meth:`get_me`.
|
||||
|
||||
Warning:
|
||||
This value is the cached return value of :meth:`get_me`. If the bots profile is
|
||||
changed during runtime, this value won't reflect the changes until :meth:`get_me` is
|
||||
called again.
|
||||
|
||||
.. seealso:: :meth:`initialize`
|
||||
"""
|
||||
if self._bot_user is None:
|
||||
raise RuntimeError(
|
||||
f"{self.__class__.__name__} is not properly initialized. Call "
|
||||
f"`{self.__class__.__name__}.initialize` before accessing this property."
|
||||
)
|
||||
return self._bot_user
|
||||
|
||||
@property
|
||||
def id(self) -> int:
|
||||
""":obj:`int`: Unique identifier for this bot. Shortcut for the corresponding attribute of
|
||||
:attr:`bot`.
|
||||
"""
|
||||
return self.bot.id
|
||||
|
||||
@property
|
||||
def first_name(self) -> str:
|
||||
""":obj:`str`: Bot's first name. Shortcut for the corresponding attribute of
|
||||
:attr:`bot`.
|
||||
"""
|
||||
return self.bot.first_name
|
||||
|
||||
@property
|
||||
def last_name(self) -> str:
|
||||
""":obj:`str`: Optional. Bot's last name. Shortcut for the corresponding attribute of
|
||||
:attr:`bot`.
|
||||
"""
|
||||
return self.bot.last_name # type: ignore
|
||||
|
||||
@property
|
||||
def username(self) -> str:
|
||||
""":obj:`str`: Bot's username. Shortcut for the corresponding attribute of
|
||||
:attr:`bot`.
|
||||
"""
|
||||
return self.bot.username # type: ignore
|
||||
|
||||
@property
|
||||
def link(self) -> str:
|
||||
""":obj:`str`: Convenience property. Returns the t.me link of the bot."""
|
||||
return f"https://t.me/{self.username}"
|
||||
|
||||
@property
|
||||
def can_join_groups(self) -> bool:
|
||||
""":obj:`bool`: Bot's :attr:`telegram.User.can_join_groups` attribute. Shortcut for the
|
||||
corresponding attribute of :attr:`bot`.
|
||||
"""
|
||||
return self.bot.can_join_groups # type: ignore
|
||||
|
||||
@property
|
||||
def can_read_all_group_messages(self) -> bool:
|
||||
""":obj:`bool`: Bot's :attr:`telegram.User.can_read_all_group_messages` attribute.
|
||||
Shortcut for the corresponding attribute of :attr:`bot`.
|
||||
"""
|
||||
return self.bot.can_read_all_group_messages # type: ignore
|
||||
|
||||
@property
|
||||
def supports_inline_queries(self) -> bool:
|
||||
""":obj:`bool`: Bot's :attr:`telegram.User.supports_inline_queries` attribute.
|
||||
Shortcut for the corresponding attribute of :attr:`bot`.
|
||||
"""
|
||||
return self.bot.supports_inline_queries # type: ignore
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
""":obj:`str`: Bot's @username. Shortcut for the corresponding attribute of :attr:`bot`."""
|
||||
return f"@{self.username}"
|
||||
|
||||
@_log
|
||||
async def get_me(
|
||||
self,
|
||||
|
|
@ -7855,14 +7863,6 @@ CUSTOM_EMOJI_IDENTIFIER_LIMIT` custom emoji identifiers can be specified.
|
|||
|
||||
return data
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if isinstance(other, self.__class__):
|
||||
return self.bot == other.bot
|
||||
return False
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash((self.__class__, self.bot))
|
||||
|
||||
# camelCase aliases
|
||||
getMe = get_me
|
||||
"""Alias for :meth:`get_me`"""
|
||||
|
|
|
|||
|
|
@ -110,43 +110,53 @@ class TelegramObject:
|
|||
# We don't do anything with api_kwargs here - see docstring of _apply_api_kwargs
|
||||
self.api_kwargs: Mapping[str, Any] = MappingProxyType(api_kwargs or {})
|
||||
|
||||
def _freeze(self) -> None:
|
||||
self._frozen = True
|
||||
def __eq__(self, other: object) -> bool:
|
||||
"""Compares this object with :paramref:`other` in terms of equality.
|
||||
If this object and :paramref:`other` are `not` objects of the same class,
|
||||
this comparison will fall back to Python's default implementation of :meth:`object.__eq__`.
|
||||
Otherwise, both objects may be compared in terms of equality, if the corresponding
|
||||
subclass of :class:`TelegramObject` has defined a set of attributes to compare and
|
||||
the objects are considered to be equal, if all of these attributes are equal.
|
||||
If the subclass has not defined a set of attributes to compare, a warning will be issued.
|
||||
|
||||
def _unfreeze(self) -> None:
|
||||
self._frozen = False
|
||||
Tip:
|
||||
If instances of a class in the :mod:`telegram` module are comparable in terms of
|
||||
equality, the documentation of the class will state the attributes that will be used
|
||||
for this comparison.
|
||||
|
||||
@contextmanager
|
||||
def _unfrozen(self: Tele_co) -> Iterator[Tele_co]:
|
||||
"""Context manager to temporarily unfreeze the object. For internal use only.
|
||||
Args:
|
||||
other (:obj:`object`): The object to compare with.
|
||||
|
||||
Returns:
|
||||
:obj:`bool`
|
||||
|
||||
Note:
|
||||
with to._unfrozen() as other_to:
|
||||
assert to is other_to
|
||||
"""
|
||||
self._unfreeze()
|
||||
yield self
|
||||
self._freeze()
|
||||
if isinstance(other, self.__class__):
|
||||
if not self._id_attrs:
|
||||
warn(
|
||||
f"Objects of type {self.__class__.__name__} can not be meaningfully tested for"
|
||||
" equivalence.",
|
||||
stacklevel=2,
|
||||
)
|
||||
if not other._id_attrs:
|
||||
warn(
|
||||
f"Objects of type {other.__class__.__name__} can not be meaningfully tested"
|
||||
" for equivalence.",
|
||||
stacklevel=2,
|
||||
)
|
||||
return self._id_attrs == other._id_attrs
|
||||
return super().__eq__(other)
|
||||
|
||||
def _apply_api_kwargs(self, api_kwargs: JSONDict) -> None:
|
||||
"""Loops through the api kwargs and for every key that exists as attribute of the
|
||||
object (and is None), it moves the value from `api_kwargs` to the attribute.
|
||||
*Edits `api_kwargs` in place!*
|
||||
def __hash__(self) -> int:
|
||||
"""Builds a hash value for this object such that the hash of two objects is equal if and
|
||||
only if the objects are equal in terms of :meth:`__eq__`.
|
||||
|
||||
This method is currently only called in the unpickling process, i.e. not on "normal" init.
|
||||
This is because
|
||||
* automating this is tricky to get right: It should be called at the *end* of the __init__,
|
||||
preferably only once at the end of the __init__ of the last child class. This could be
|
||||
done via __init_subclass__, but it's hard to not destroy the signature of __init__ in the
|
||||
process.
|
||||
* calling it manually in every __init__ is tedious
|
||||
* There probably is no use case for it anyway. If you manually initialize a TO subclass,
|
||||
then you can pass everything as proper argument.
|
||||
Returns:
|
||||
:obj:`int`
|
||||
"""
|
||||
# we convert to list to ensure that the list doesn't change length while we loop
|
||||
for key in list(api_kwargs.keys()):
|
||||
if getattr(self, key, True) is None:
|
||||
setattr(self, key, api_kwargs.pop(key))
|
||||
if self._id_attrs:
|
||||
return hash((self.__class__, self._id_attrs))
|
||||
return super().__hash__()
|
||||
|
||||
def __setattr__(self, key: str, value: object) -> None:
|
||||
"""Overrides :meth:`object.__setattr__` to prevent the overriding of attributes.
|
||||
|
|
@ -364,6 +374,122 @@ class TelegramObject:
|
|||
self.set_bot(bot)
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def _parse_data(data: Optional[JSONDict]) -> Optional[JSONDict]:
|
||||
"""Should be called by subclasses that override de_json to ensure that the input
|
||||
is not altered. Whoever calls de_json might still want to use the original input
|
||||
for something else.
|
||||
"""
|
||||
return None if data is None else data.copy()
|
||||
|
||||
@classmethod
|
||||
def _de_json(
|
||||
cls: Type[Tele_co],
|
||||
data: Optional[JSONDict],
|
||||
bot: "Bot",
|
||||
api_kwargs: Optional[JSONDict] = None,
|
||||
) -> Optional[Tele_co]:
|
||||
if data is None:
|
||||
return None
|
||||
|
||||
# try-except is significantly faster in case we already have a correct argument set
|
||||
try:
|
||||
obj = cls(**data, api_kwargs=api_kwargs)
|
||||
except TypeError as exc:
|
||||
if "__init__() got an unexpected keyword argument" not in str(exc):
|
||||
raise exc
|
||||
|
||||
if cls.__INIT_PARAMS_CHECK is not cls:
|
||||
signature = inspect.signature(cls)
|
||||
cls.__INIT_PARAMS = set(signature.parameters.keys())
|
||||
cls.__INIT_PARAMS_CHECK = cls
|
||||
|
||||
api_kwargs = api_kwargs or {}
|
||||
existing_kwargs: JSONDict = {}
|
||||
for key, value in data.items():
|
||||
(existing_kwargs if key in cls.__INIT_PARAMS else api_kwargs)[key] = value
|
||||
|
||||
obj = cls(api_kwargs=api_kwargs, **existing_kwargs)
|
||||
|
||||
obj.set_bot(bot=bot)
|
||||
return obj
|
||||
|
||||
@classmethod
|
||||
def de_json(cls: Type[Tele_co], data: Optional[JSONDict], bot: "Bot") -> Optional[Tele_co]:
|
||||
"""Converts JSON data to a Telegram object.
|
||||
|
||||
Args:
|
||||
data (Dict[:obj:`str`, ...]): The JSON data.
|
||||
bot (:class:`telegram.Bot`): The bot associated with this object.
|
||||
|
||||
Returns:
|
||||
The Telegram object.
|
||||
|
||||
"""
|
||||
return cls._de_json(data=data, bot=bot)
|
||||
|
||||
@classmethod
|
||||
def de_list(
|
||||
cls: Type[Tele_co], data: Optional[List[JSONDict]], bot: "Bot"
|
||||
) -> Tuple[Tele_co, ...]:
|
||||
"""Converts a list of JSON objects to a tuple of Telegram objects.
|
||||
|
||||
.. versionchanged:: 20.0
|
||||
|
||||
* Returns a tuple instead of a list.
|
||||
* Filters out any :obj:`None` values.
|
||||
|
||||
Args:
|
||||
data (List[Dict[:obj:`str`, ...]]): The JSON data.
|
||||
bot (:class:`telegram.Bot`): The bot associated with these objects.
|
||||
|
||||
Returns:
|
||||
A tuple of Telegram objects.
|
||||
|
||||
"""
|
||||
if not data:
|
||||
return ()
|
||||
|
||||
return tuple(obj for obj in (cls.de_json(d, bot) for d in data) if obj is not None)
|
||||
|
||||
@contextmanager
|
||||
def _unfrozen(self: Tele_co) -> Iterator[Tele_co]:
|
||||
"""Context manager to temporarily unfreeze the object. For internal use only.
|
||||
|
||||
Note:
|
||||
with to._unfrozen() as other_to:
|
||||
assert to is other_to
|
||||
"""
|
||||
self._unfreeze()
|
||||
yield self
|
||||
self._freeze()
|
||||
|
||||
def _freeze(self) -> None:
|
||||
self._frozen = True
|
||||
|
||||
def _unfreeze(self) -> None:
|
||||
self._frozen = False
|
||||
|
||||
def _apply_api_kwargs(self, api_kwargs: JSONDict) -> None:
|
||||
"""Loops through the api kwargs and for every key that exists as attribute of the
|
||||
object (and is None), it moves the value from `api_kwargs` to the attribute.
|
||||
*Edits `api_kwargs` in place!*
|
||||
|
||||
This method is currently only called in the unpickling process, i.e. not on "normal" init.
|
||||
This is because
|
||||
* automating this is tricky to get right: It should be called at the *end* of the __init__,
|
||||
preferably only once at the end of the __init__ of the last child class. This could be
|
||||
done via __init_subclass__, but it's hard to not destroy the signature of __init__ in the
|
||||
process.
|
||||
* calling it manually in every __init__ is tedious
|
||||
* There probably is no use case for it anyway. If you manually initialize a TO subclass,
|
||||
then you can pass everything as proper argument.
|
||||
"""
|
||||
# we convert to list to ensure that the list doesn't change length while we loop
|
||||
for key in list(api_kwargs.keys()):
|
||||
if getattr(self, key, True) is None:
|
||||
setattr(self, key, api_kwargs.pop(key))
|
||||
|
||||
def _get_attrs_names(self, include_private: bool) -> Iterator[str]:
|
||||
"""
|
||||
Returns the names of the attributes of this object. This is used to determine which
|
||||
|
|
@ -423,84 +549,6 @@ class TelegramObject:
|
|||
data.pop("_bot", None)
|
||||
return data
|
||||
|
||||
@staticmethod
|
||||
def _parse_data(data: Optional[JSONDict]) -> Optional[JSONDict]:
|
||||
"""Should be called by subclasses that override de_json to ensure that the input
|
||||
is not altered. Whoever calls de_json might still want to use the original input
|
||||
for something else.
|
||||
"""
|
||||
return None if data is None else data.copy()
|
||||
|
||||
@classmethod
|
||||
def de_json(cls: Type[Tele_co], data: Optional[JSONDict], bot: "Bot") -> Optional[Tele_co]:
|
||||
"""Converts JSON data to a Telegram object.
|
||||
|
||||
Args:
|
||||
data (Dict[:obj:`str`, ...]): The JSON data.
|
||||
bot (:class:`telegram.Bot`): The bot associated with this object.
|
||||
|
||||
Returns:
|
||||
The Telegram object.
|
||||
|
||||
"""
|
||||
return cls._de_json(data=data, bot=bot)
|
||||
|
||||
@classmethod
|
||||
def _de_json(
|
||||
cls: Type[Tele_co],
|
||||
data: Optional[JSONDict],
|
||||
bot: "Bot",
|
||||
api_kwargs: Optional[JSONDict] = None,
|
||||
) -> Optional[Tele_co]:
|
||||
if data is None:
|
||||
return None
|
||||
|
||||
# try-except is significantly faster in case we already have a correct argument set
|
||||
try:
|
||||
obj = cls(**data, api_kwargs=api_kwargs)
|
||||
except TypeError as exc:
|
||||
if "__init__() got an unexpected keyword argument" not in str(exc):
|
||||
raise exc
|
||||
|
||||
if cls.__INIT_PARAMS_CHECK is not cls:
|
||||
signature = inspect.signature(cls)
|
||||
cls.__INIT_PARAMS = set(signature.parameters.keys())
|
||||
cls.__INIT_PARAMS_CHECK = cls
|
||||
|
||||
api_kwargs = api_kwargs or {}
|
||||
existing_kwargs: JSONDict = {}
|
||||
for key, value in data.items():
|
||||
(existing_kwargs if key in cls.__INIT_PARAMS else api_kwargs)[key] = value
|
||||
|
||||
obj = cls(api_kwargs=api_kwargs, **existing_kwargs)
|
||||
|
||||
obj.set_bot(bot=bot)
|
||||
return obj
|
||||
|
||||
@classmethod
|
||||
def de_list(
|
||||
cls: Type[Tele_co], data: Optional[List[JSONDict]], bot: "Bot"
|
||||
) -> Tuple[Tele_co, ...]:
|
||||
"""Converts a list of JSON objects to a tuple of Telegram objects.
|
||||
|
||||
.. versionchanged:: 20.0
|
||||
|
||||
* Returns a tuple instead of a list.
|
||||
* Filters out any :obj:`None` values.
|
||||
|
||||
Args:
|
||||
data (List[Dict[:obj:`str`, ...]]): The JSON data.
|
||||
bot (:class:`telegram.Bot`): The bot associated with these objects.
|
||||
|
||||
Returns:
|
||||
A tuple of Telegram objects.
|
||||
|
||||
"""
|
||||
if not data:
|
||||
return ()
|
||||
|
||||
return tuple(obj for obj in (cls.de_json(d, bot) for d in data) if obj is not None)
|
||||
|
||||
def to_json(self) -> str:
|
||||
"""Gives a JSON representation of object.
|
||||
|
||||
|
|
@ -596,51 +644,3 @@ class TelegramObject:
|
|||
bot (:class:`telegram.Bot` | :obj:`None`): The bot instance.
|
||||
"""
|
||||
self._bot = bot
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
"""Compares this object with :paramref:`other` in terms of equality.
|
||||
If this object and :paramref:`other` are `not` objects of the same class,
|
||||
this comparison will fall back to Python's default implementation of :meth:`object.__eq__`.
|
||||
Otherwise, both objects may be compared in terms of equality, if the corresponding
|
||||
subclass of :class:`TelegramObject` has defined a set of attributes to compare and
|
||||
the objects are considered to be equal, if all of these attributes are equal.
|
||||
If the subclass has not defined a set of attributes to compare, a warning will be issued.
|
||||
|
||||
Tip:
|
||||
If instances of a class in the :mod:`telegram` module are comparable in terms of
|
||||
equality, the documentation of the class will state the attributes that will be used
|
||||
for this comparison.
|
||||
|
||||
Args:
|
||||
other (:obj:`object`): The object to compare with.
|
||||
|
||||
Returns:
|
||||
:obj:`bool`
|
||||
|
||||
"""
|
||||
if isinstance(other, self.__class__):
|
||||
if not self._id_attrs:
|
||||
warn(
|
||||
f"Objects of type {self.__class__.__name__} can not be meaningfully tested for"
|
||||
" equivalence.",
|
||||
stacklevel=2,
|
||||
)
|
||||
if not other._id_attrs:
|
||||
warn(
|
||||
f"Objects of type {other.__class__.__name__} can not be meaningfully tested"
|
||||
" for equivalence.",
|
||||
stacklevel=2,
|
||||
)
|
||||
return self._id_attrs == other._id_attrs
|
||||
return super().__eq__(other)
|
||||
|
||||
def __hash__(self) -> int:
|
||||
"""Builds a hash value for this object such that the hash of two objects is equal if and
|
||||
only if the objects are equal in terms of :meth:`__eq__`.
|
||||
|
||||
Returns:
|
||||
:obj:`int`
|
||||
"""
|
||||
if self._id_attrs:
|
||||
return hash((self.__class__, self._id_attrs))
|
||||
return super().__hash__()
|
||||
|
|
|
|||
|
|
@ -88,6 +88,14 @@ class DefaultValue(Generic[DVType]):
|
|||
def __bool__(self) -> bool:
|
||||
return bool(self.value)
|
||||
|
||||
# This is mostly here for readability during debugging
|
||||
def __str__(self) -> str:
|
||||
return f"DefaultValue({self.value})"
|
||||
|
||||
# This is here to have the default instances nicely rendered in the docs
|
||||
def __repr__(self) -> str:
|
||||
return repr(self.value)
|
||||
|
||||
@overload
|
||||
@staticmethod
|
||||
def get_value(obj: "DefaultValue[OT]") -> OT:
|
||||
|
|
@ -112,14 +120,6 @@ class DefaultValue(Generic[DVType]):
|
|||
"""
|
||||
return obj.value if isinstance(obj, DefaultValue) else obj
|
||||
|
||||
# This is mostly here for readability during debugging
|
||||
def __str__(self) -> str:
|
||||
return f"DefaultValue({self.value})"
|
||||
|
||||
# This is here to have the default instances nicely rendered in the docs
|
||||
def __repr__(self) -> str:
|
||||
return repr(self.value)
|
||||
|
||||
|
||||
DEFAULT_NONE: DefaultValue[None] = DefaultValue(None)
|
||||
""":class:`DefaultValue`: Default :obj:`None`"""
|
||||
|
|
|
|||
|
|
@ -344,6 +344,26 @@ class Application(Generic[BT, CCT, UD, CD, BD, JQ], AsyncContextManager["Applica
|
|||
self.__update_persistence_lock = asyncio.Lock()
|
||||
self.__create_task_tasks: Set[asyncio.Task] = set() # Used for awaiting tasks upon exit
|
||||
|
||||
async def __aenter__(self: _AppType) -> _AppType: # noqa: PYI019
|
||||
"""Simple context manager which initializes the App."""
|
||||
try:
|
||||
await self.initialize()
|
||||
return self
|
||||
except Exception as exc:
|
||||
await self.shutdown()
|
||||
raise exc
|
||||
|
||||
async def __aexit__(
|
||||
self,
|
||||
exc_type: Optional[Type[BaseException]],
|
||||
exc_val: Optional[BaseException],
|
||||
exc_tb: Optional[TracebackType],
|
||||
) -> None:
|
||||
"""Shutdown the App from the context manager."""
|
||||
# Make sure not to return `True` so that exceptions are not suppressed
|
||||
# https://docs.python.org/3/reference/datamodel.html?#object.__aexit__
|
||||
await self.shutdown()
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""Give a string representation of the application in the form ``Application[bot=...]``.
|
||||
|
||||
|
|
@ -355,12 +375,6 @@ class Application(Generic[BT, CCT, UD, CD, BD, JQ], AsyncContextManager["Applica
|
|||
"""
|
||||
return build_repr_with_selected_attrs(self, bot=self.bot)
|
||||
|
||||
def _check_initialized(self) -> None:
|
||||
if not self._initialized:
|
||||
raise RuntimeError(
|
||||
"This Application was not initialized via `Application.initialize`!"
|
||||
)
|
||||
|
||||
@property
|
||||
def running(self) -> bool:
|
||||
""":obj:`bool`: Indicates if this application is running.
|
||||
|
|
@ -410,6 +424,27 @@ class Application(Generic[BT, CCT, UD, CD, BD, JQ], AsyncContextManager["Applica
|
|||
"""
|
||||
return self._update_processor
|
||||
|
||||
@staticmethod
|
||||
def _raise_system_exit() -> NoReturn:
|
||||
raise SystemExit
|
||||
|
||||
@staticmethod
|
||||
def builder() -> "InitApplicationBuilder":
|
||||
"""Convenience method. Returns a new :class:`telegram.ext.ApplicationBuilder`.
|
||||
|
||||
.. versionadded:: 20.0
|
||||
"""
|
||||
# Unfortunately this needs to be here due to cyclical imports
|
||||
from telegram.ext import ApplicationBuilder # pylint: disable=import-outside-toplevel
|
||||
|
||||
return ApplicationBuilder()
|
||||
|
||||
def _check_initialized(self) -> None:
|
||||
if not self._initialized:
|
||||
raise RuntimeError(
|
||||
"This Application was not initialized via `Application.initialize`!"
|
||||
)
|
||||
|
||||
async def initialize(self) -> None:
|
||||
"""Initializes the Application by initializing:
|
||||
|
||||
|
|
@ -496,26 +531,6 @@ class Application(Generic[BT, CCT, UD, CD, BD, JQ], AsyncContextManager["Applica
|
|||
|
||||
self._initialized = False
|
||||
|
||||
async def __aenter__(self: _AppType) -> _AppType: # noqa: PYI019
|
||||
"""Simple context manager which initializes the App."""
|
||||
try:
|
||||
await self.initialize()
|
||||
return self
|
||||
except Exception as exc:
|
||||
await self.shutdown()
|
||||
raise exc
|
||||
|
||||
async def __aexit__(
|
||||
self,
|
||||
exc_type: Optional[Type[BaseException]],
|
||||
exc_val: Optional[BaseException],
|
||||
exc_tb: Optional[TracebackType],
|
||||
) -> None:
|
||||
"""Shutdown the App from the context manager."""
|
||||
# Make sure not to return `True` so that exceptions are not suppressed
|
||||
# https://docs.python.org/3/reference/datamodel.html?#object.__aexit__
|
||||
await self.shutdown()
|
||||
|
||||
async def _initialize_persistence(self) -> None:
|
||||
"""This method basically just loads all the data by awaiting the BP methods"""
|
||||
if not self.persistence:
|
||||
|
|
@ -545,17 +560,6 @@ class Application(Generic[BT, CCT, UD, CD, BD, JQ], AsyncContextManager["Applica
|
|||
persistent_data
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def builder() -> "InitApplicationBuilder":
|
||||
"""Convenience method. Returns a new :class:`telegram.ext.ApplicationBuilder`.
|
||||
|
||||
.. versionadded:: 20.0
|
||||
"""
|
||||
# Unfortunately this needs to be here due to cyclical imports
|
||||
from telegram.ext import ApplicationBuilder # pylint: disable=import-outside-toplevel
|
||||
|
||||
return ApplicationBuilder()
|
||||
|
||||
async def start(self) -> None:
|
||||
"""Starts
|
||||
|
||||
|
|
@ -922,10 +926,6 @@ class Application(Generic[BT, CCT, UD, CD, BD, JQ], AsyncContextManager["Applica
|
|||
stop_signals=stop_signals,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _raise_system_exit() -> NoReturn:
|
||||
raise SystemExit
|
||||
|
||||
def __run(
|
||||
self,
|
||||
updater_coroutine: Coroutine,
|
||||
|
|
|
|||
|
|
@ -48,6 +48,24 @@ class BaseUpdateProcessor(ABC):
|
|||
raise ValueError("`max_concurrent_updates` must be a positive integer!")
|
||||
self._semaphore = BoundedSemaphore(self.max_concurrent_updates)
|
||||
|
||||
async def __aenter__(self) -> "BaseUpdateProcessor":
|
||||
"""Simple context manager which initializes the Processor."""
|
||||
try:
|
||||
await self.initialize()
|
||||
return self
|
||||
except Exception as exc:
|
||||
await self.shutdown()
|
||||
raise exc
|
||||
|
||||
async def __aexit__(
|
||||
self,
|
||||
exc_type: Optional[Type[BaseException]],
|
||||
exc_val: Optional[BaseException],
|
||||
exc_tb: Optional[TracebackType],
|
||||
) -> None:
|
||||
"""Simple context manager which shuts down the Processor."""
|
||||
await self.shutdown()
|
||||
|
||||
@property
|
||||
def max_concurrent_updates(self) -> int:
|
||||
""":obj:`int`: The maximum number of updates that can be processed concurrently."""
|
||||
|
|
@ -105,24 +123,6 @@ class BaseUpdateProcessor(ABC):
|
|||
async with self._semaphore:
|
||||
await self.do_process_update(update, coroutine)
|
||||
|
||||
async def __aenter__(self) -> "BaseUpdateProcessor":
|
||||
"""Simple context manager which initializes the Processor."""
|
||||
try:
|
||||
await self.initialize()
|
||||
return self
|
||||
except Exception as exc:
|
||||
await self.shutdown()
|
||||
raise exc
|
||||
|
||||
async def __aexit__(
|
||||
self,
|
||||
exc_type: Optional[Type[BaseException]],
|
||||
exc_val: Optional[BaseException],
|
||||
exc_tb: Optional[TracebackType],
|
||||
) -> None:
|
||||
"""Shutdown the Processor from the context manager."""
|
||||
await self.shutdown()
|
||||
|
||||
|
||||
class SimpleUpdateProcessor(BaseUpdateProcessor):
|
||||
"""Instance of :class:`telegram.ext.BaseUpdateProcessor` that immediately awaits the
|
||||
|
|
|
|||
|
|
@ -104,6 +104,25 @@ class Defaults:
|
|||
if value is not None:
|
||||
self._api_defaults[kwarg] = value
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(
|
||||
(
|
||||
self._parse_mode,
|
||||
self._disable_notification,
|
||||
self._disable_web_page_preview,
|
||||
self._allow_sending_without_reply,
|
||||
self._quote,
|
||||
self._tzinfo,
|
||||
self._block,
|
||||
self._protect_content,
|
||||
)
|
||||
)
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if isinstance(other, Defaults):
|
||||
return all(getattr(self, attr) == getattr(other, attr) for attr in self.__slots__)
|
||||
return False
|
||||
|
||||
@property
|
||||
def api_defaults(self) -> Dict[str, Any]: # skip-cq: PY-D0003
|
||||
return self._api_defaults
|
||||
|
|
@ -220,22 +239,3 @@ class Defaults:
|
|||
raise AttributeError(
|
||||
"You can't assign a new value to protect_content after initialization."
|
||||
)
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(
|
||||
(
|
||||
self._parse_mode,
|
||||
self._disable_notification,
|
||||
self._disable_web_page_preview,
|
||||
self._allow_sending_without_reply,
|
||||
self._quote,
|
||||
self._tzinfo,
|
||||
self._block,
|
||||
self._protect_content,
|
||||
)
|
||||
)
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if isinstance(other, Defaults):
|
||||
return all(getattr(self, attr) == getattr(other, attr) for attr in self.__slots__)
|
||||
return False
|
||||
|
|
|
|||
|
|
@ -109,6 +109,16 @@ class JobQueue(Generic[CCT]):
|
|||
"""
|
||||
return build_repr_with_selected_attrs(self, application=self.application)
|
||||
|
||||
@property
|
||||
def application(self) -> "Application[Any, CCT, Any, Any, Any, JobQueue[CCT]]":
|
||||
"""The application this JobQueue is associated with."""
|
||||
if self._application is None:
|
||||
raise RuntimeError("No application was set for this JobQueue.")
|
||||
application = self._application()
|
||||
if application is not None:
|
||||
return application
|
||||
raise RuntimeError("The application instance is no longer alive.")
|
||||
|
||||
def _tz_now(self) -> datetime.datetime:
|
||||
return datetime.datetime.now(self.scheduler.timezone)
|
||||
|
||||
|
|
@ -162,16 +172,6 @@ class JobQueue(Generic[CCT]):
|
|||
executors={"default": self._executor},
|
||||
)
|
||||
|
||||
@property
|
||||
def application(self) -> "Application[Any, CCT, Any, Any, Any, JobQueue[CCT]]":
|
||||
"""The application this JobQueue is associated with."""
|
||||
if self._application is None:
|
||||
raise RuntimeError("No application was set for this JobQueue.")
|
||||
application = self._application()
|
||||
if application is not None:
|
||||
return application
|
||||
raise RuntimeError("The application instance is no longer alive.")
|
||||
|
||||
@staticmethod
|
||||
async def job_callback(job_queue: "JobQueue[CCT]", job: "Job[CCT]") -> None:
|
||||
"""This method is used as a callback for the APScheduler jobs.
|
||||
|
|
@ -778,6 +778,22 @@ class Job(Generic[CCT]):
|
|||
|
||||
self._job = cast("APSJob", None) # skipcq: PTC-W0052
|
||||
|
||||
def __getattr__(self, item: str) -> object:
|
||||
try:
|
||||
return getattr(self.job, item)
|
||||
except AttributeError as exc:
|
||||
raise AttributeError(
|
||||
f"Neither 'telegram.ext.Job' nor 'apscheduler.job.Job' has attribute '{item}'"
|
||||
) from exc
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if isinstance(other, self.__class__):
|
||||
return self.id == other.id
|
||||
return False
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(self.id)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""Give a string representation of the job in the form
|
||||
``Job[id=..., name=..., callback=..., trigger=...]``.
|
||||
|
|
@ -805,46 +821,6 @@ class Job(Generic[CCT]):
|
|||
"""
|
||||
return self._job
|
||||
|
||||
async def run(
|
||||
self, application: "Application[Any, CCT, Any, Any, Any, JobQueue[CCT]]"
|
||||
) -> None:
|
||||
"""Executes the callback function independently of the jobs schedule. Also calls
|
||||
:meth:`telegram.ext.Application.update_persistence`.
|
||||
|
||||
.. versionchanged:: 20.0
|
||||
Calls :meth:`telegram.ext.Application.update_persistence`.
|
||||
|
||||
Args:
|
||||
application (:class:`telegram.ext.Application`): The application this job is associated
|
||||
with.
|
||||
"""
|
||||
# We shield the task such that the job isn't cancelled mid-run
|
||||
await asyncio.shield(self._run(application))
|
||||
|
||||
async def _run(
|
||||
self, application: "Application[Any, CCT, Any, Any, Any, JobQueue[CCT]]"
|
||||
) -> None:
|
||||
try:
|
||||
context = application.context_types.context.from_job(self, application)
|
||||
await context.refresh_data()
|
||||
await self.callback(context)
|
||||
except Exception as exc:
|
||||
await application.create_task(
|
||||
application.process_error(None, exc, job=self),
|
||||
name=f"Job:{self.id}:run:process_error",
|
||||
)
|
||||
finally:
|
||||
# This is internal logic of application - let's keep it private for now
|
||||
application._mark_for_persistence_update(job=self) # pylint: disable=protected-access
|
||||
|
||||
def schedule_removal(self) -> None:
|
||||
"""
|
||||
Schedules this job for removal from the :class:`JobQueue`. It will be removed without
|
||||
executing its callback function again.
|
||||
"""
|
||||
self.job.remove()
|
||||
self._removed = True
|
||||
|
||||
@property
|
||||
def removed(self) -> bool:
|
||||
""":obj:`bool`: Whether this job is due to be removed."""
|
||||
|
|
@ -897,18 +873,42 @@ class Job(Generic[CCT]):
|
|||
ext_job._job = aps_job # pylint: disable=protected-access
|
||||
return ext_job
|
||||
|
||||
def __getattr__(self, item: str) -> object:
|
||||
async def run(
|
||||
self, application: "Application[Any, CCT, Any, Any, Any, JobQueue[CCT]]"
|
||||
) -> None:
|
||||
"""Executes the callback function independently of the jobs schedule. Also calls
|
||||
:meth:`telegram.ext.Application.update_persistence`.
|
||||
|
||||
.. versionchanged:: 20.0
|
||||
Calls :meth:`telegram.ext.Application.update_persistence`.
|
||||
|
||||
Args:
|
||||
application (:class:`telegram.ext.Application`): The application this job is associated
|
||||
with.
|
||||
"""
|
||||
# We shield the task such that the job isn't cancelled mid-run
|
||||
await asyncio.shield(self._run(application))
|
||||
|
||||
async def _run(
|
||||
self, application: "Application[Any, CCT, Any, Any, Any, JobQueue[CCT]]"
|
||||
) -> None:
|
||||
try:
|
||||
return getattr(self.job, item)
|
||||
except AttributeError as exc:
|
||||
raise AttributeError(
|
||||
f"Neither 'telegram.ext.Job' nor 'apscheduler.job.Job' has attribute '{item}'"
|
||||
) from exc
|
||||
context = application.context_types.context.from_job(self, application)
|
||||
await context.refresh_data()
|
||||
await self.callback(context)
|
||||
except Exception as exc:
|
||||
await application.create_task(
|
||||
application.process_error(None, exc, job=self),
|
||||
name=f"Job:{self.id}:run:process_error",
|
||||
)
|
||||
finally:
|
||||
# This is internal logic of application - let's keep it private for now
|
||||
application._mark_for_persistence_update(job=self) # pylint: disable=protected-access
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if isinstance(other, self.__class__):
|
||||
return self.id == other.id
|
||||
return False
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(self.id)
|
||||
def schedule_removal(self) -> None:
|
||||
"""
|
||||
Schedules this job for removal from the :class:`JobQueue`. It will be removed without
|
||||
executing its callback function again.
|
||||
"""
|
||||
self.job.remove()
|
||||
self._removed = True
|
||||
|
|
|
|||
|
|
@ -125,6 +125,26 @@ class Updater(AsyncContextManager["Updater"]):
|
|||
self.__polling_task: Optional[asyncio.Task] = None
|
||||
self.__polling_cleanup_cb: Optional[Callable[[], Coroutine[Any, Any, None]]] = None
|
||||
|
||||
async def __aenter__(self: _UpdaterType) -> _UpdaterType: # noqa: PYI019
|
||||
"""Simple context manager which initializes the Updater."""
|
||||
try:
|
||||
await self.initialize()
|
||||
return self
|
||||
except Exception as exc:
|
||||
await self.shutdown()
|
||||
raise exc
|
||||
|
||||
async def __aexit__(
|
||||
self,
|
||||
exc_type: Optional[Type[BaseException]],
|
||||
exc_val: Optional[BaseException],
|
||||
exc_tb: Optional[TracebackType],
|
||||
) -> None:
|
||||
"""Shutdown the Updater from the context manager."""
|
||||
# Make sure not to return `True` so that exceptions are not suppressed
|
||||
# https://docs.python.org/3/reference/datamodel.html?#object.__aexit__
|
||||
await self.shutdown()
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""Give a string representation of the updater in the form ``Updater[bot=...]``.
|
||||
|
||||
|
|
@ -175,26 +195,6 @@ class Updater(AsyncContextManager["Updater"]):
|
|||
self._initialized = False
|
||||
_LOGGER.debug("Shut down of Updater complete")
|
||||
|
||||
async def __aenter__(self: _UpdaterType) -> _UpdaterType: # noqa: PYI019
|
||||
"""Simple context manager which initializes the Updater."""
|
||||
try:
|
||||
await self.initialize()
|
||||
return self
|
||||
except Exception as exc:
|
||||
await self.shutdown()
|
||||
raise exc
|
||||
|
||||
async def __aexit__(
|
||||
self,
|
||||
exc_type: Optional[Type[BaseException]],
|
||||
exc_val: Optional[BaseException],
|
||||
exc_tb: Optional[TracebackType],
|
||||
) -> None:
|
||||
"""Shutdown the Updater from the context manager."""
|
||||
# Make sure not to return `True` so that exceptions are not suppressed
|
||||
# https://docs.python.org/3/reference/datamodel.html?#object.__aexit__
|
||||
await self.shutdown()
|
||||
|
||||
async def start_polling(
|
||||
self,
|
||||
poll_interval: float = 0.0,
|
||||
|
|
|
|||
|
|
@ -54,6 +54,14 @@ class TrackingDict(UserDict, Generic[_KT, _VT]):
|
|||
super().__init__()
|
||||
self._write_access_keys: Set[_KT] = set()
|
||||
|
||||
def __setitem__(self, key: _KT, value: _VT) -> None:
|
||||
self.__track_write(key)
|
||||
super().__setitem__(key, value)
|
||||
|
||||
def __delitem__(self, key: _KT) -> None:
|
||||
self.__track_write(key)
|
||||
super().__delitem__(key)
|
||||
|
||||
def __track_write(self, key: Union[_KT, Set[_KT]]) -> None:
|
||||
if isinstance(key, set):
|
||||
self._write_access_keys |= key
|
||||
|
|
@ -83,14 +91,6 @@ class TrackingDict(UserDict, Generic[_KT, _VT]):
|
|||
|
||||
# Override methods to track access
|
||||
|
||||
def __setitem__(self, key: _KT, value: _VT) -> None:
|
||||
self.__track_write(key)
|
||||
super().__setitem__(key, value)
|
||||
|
||||
def __delitem__(self, key: _KT) -> None:
|
||||
self.__track_write(key)
|
||||
super().__delitem__(key)
|
||||
|
||||
def update_no_track(self, mapping: Mapping[_KT, _VT]) -> None:
|
||||
"""Like ``update``, but doesn't count towards write access."""
|
||||
for key, value in mapping.items():
|
||||
|
|
|
|||
|
|
@ -182,6 +182,39 @@ class BaseFilter:
|
|||
self._name = self.__class__.__name__ if name is None else name
|
||||
self._data_filter = data_filter
|
||||
|
||||
def __and__(self, other: "BaseFilter") -> "BaseFilter":
|
||||
return _MergedFilter(self, and_filter=other)
|
||||
|
||||
def __or__(self, other: "BaseFilter") -> "BaseFilter":
|
||||
return _MergedFilter(self, or_filter=other)
|
||||
|
||||
def __xor__(self, other: "BaseFilter") -> "BaseFilter":
|
||||
return _XORFilter(self, other)
|
||||
|
||||
def __invert__(self) -> "BaseFilter":
|
||||
return _InvertedFilter(self)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return self.name
|
||||
|
||||
@property
|
||||
def data_filter(self) -> bool:
|
||||
""":obj:`bool`: Whether this filter is a data filter."""
|
||||
return self._data_filter
|
||||
|
||||
@data_filter.setter
|
||||
def data_filter(self, value: bool) -> None:
|
||||
self._data_filter = value
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
""":obj:`str`: Name for this filter."""
|
||||
return self._name
|
||||
|
||||
@name.setter
|
||||
def name(self, name: str) -> None:
|
||||
self._name = name
|
||||
|
||||
def check_update( # skipcq: PYL-R0201
|
||||
self, update: Update
|
||||
) -> Optional[Union[bool, FilterDataDict]]:
|
||||
|
|
@ -205,39 +238,6 @@ class BaseFilter:
|
|||
return True
|
||||
return False
|
||||
|
||||
def __and__(self, other: "BaseFilter") -> "BaseFilter":
|
||||
return _MergedFilter(self, and_filter=other)
|
||||
|
||||
def __or__(self, other: "BaseFilter") -> "BaseFilter":
|
||||
return _MergedFilter(self, or_filter=other)
|
||||
|
||||
def __xor__(self, other: "BaseFilter") -> "BaseFilter":
|
||||
return _XORFilter(self, other)
|
||||
|
||||
def __invert__(self) -> "BaseFilter":
|
||||
return _InvertedFilter(self)
|
||||
|
||||
@property
|
||||
def data_filter(self) -> bool:
|
||||
""":obj:`bool`: Whether this filter is a data filter."""
|
||||
return self._data_filter
|
||||
|
||||
@data_filter.setter
|
||||
def data_filter(self, value: bool) -> None:
|
||||
self._data_filter = value
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
""":obj:`str`: Name for this filter."""
|
||||
return self._name
|
||||
|
||||
@name.setter
|
||||
def name(self, name: str) -> None:
|
||||
self._name = name
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return self.name
|
||||
|
||||
|
||||
class MessageFilter(BaseFilter):
|
||||
"""Base class for all Message Filters. In contrast to :class:`UpdateFilter`, the object passed
|
||||
|
|
|
|||
Loading…
Reference in a new issue