Add String Representation for Selected Classes (#3826)

This commit is contained in:
Dmitry K 2023-09-15 22:35:45 +03:00 committed by GitHub
parent 39abf838fa
commit 9c7298c17a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 240 additions and 2 deletions

View file

@ -92,6 +92,7 @@ from telegram._utils.argumentparsing import parse_sequence_arg
from telegram._utils.defaultvalue import DEFAULT_NONE, DefaultValue
from telegram._utils.files import is_local_file, parse_file_input
from telegram._utils.logging import get_logger
from telegram._utils.repr import build_repr_with_selected_attrs
from telegram._utils.types import (
CorrectOptionID,
DVInput,
@ -308,6 +309,17 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
self._freeze()
def __repr__(self) -> str:
"""Give a string representation of the bot in the form ``Bot[token=...]``.
As this class doesn't implement :meth:`object.__str__`, the default implementation
will be used, which is equivalent to :meth:`__repr__`.
Returns:
:obj:`str`
"""
return build_repr_with_selected_attrs(self, token=self.token)
@property
def token(self) -> str:
""":obj:`str`: Bot's unique authentication token.

45
telegram/_utils/repr.py Normal file
View file

@ -0,0 +1,45 @@
#!/usr/bin/env python
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015-2023
# Leandro Toledo de Souza <devs@python-telegram-bot.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser Public License for more details.
#
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
"""This module contains auxiliary functionality for building strings for __repr__ method.
Warning:
Contents of this module are intended to be used internally by the library and *not* by the
user. Changes to this module are not considered breaking changes and may not be documented in
the changelog.
"""
from typing import Any
def build_repr_with_selected_attrs(obj: object, **kwargs: Any) -> str:
"""Create ``__repr__`` string in the style ``Classname[arg1=1, arg2=2]``.
The square brackets emphasize the fact that an object cannot be instantiated
from this string.
Attributes that are to be used in the representation, are passed as kwargs.
"""
return (
f"{obj.__class__.__name__}"
# square brackets emphasize that an object cannot be instantiated with these params
f"[{', '.join(_stringify(name, value) for name, value in kwargs.items())}]"
)
def _stringify(key: str, val: Any) -> str:
return f"{key}={val.__qualname__ if callable(val) else val}"

View file

@ -54,6 +54,7 @@ from typing import (
from telegram._update import Update
from telegram._utils.defaultvalue import DEFAULT_NONE, DEFAULT_TRUE, DefaultValue
from telegram._utils.logging import get_logger
from telegram._utils.repr import build_repr_with_selected_attrs
from telegram._utils.types import SCT, DVType, ODVInput
from telegram._utils.warnings import warn
from telegram.error import TelegramError
@ -80,7 +81,6 @@ _AppType = TypeVar("_AppType", bound="Application") # pylint: disable=invalid-n
_STOP_SIGNAL = object()
_DEFAULT_0 = DefaultValue(0)
# Since python 3.12, the coroutine passed to create_task should not be an (async) generator. Remove
# this check when we drop support for python 3.11.
if sys.version_info >= (3, 12):
@ -90,7 +90,6 @@ else:
_ErrorCoroType = Optional[_CoroType[RT]]
_LOGGER = get_logger(__name__)
@ -345,6 +344,17 @@ 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
def __repr__(self) -> str:
"""Give a string representation of the application in the form ``Application[bot=...]``.
As this class doesn't implement :meth:`object.__str__`, the default implementation
will be used, which is equivalent to :meth:`__repr__`.
Returns:
:obj:`str`
"""
return build_repr_with_selected_attrs(self, bot=self.bot)
def _check_initialized(self) -> None:
if not self._initialized:
raise RuntimeError(

View file

@ -21,6 +21,7 @@ from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any, Generic, Optional, TypeVar, Union
from telegram._utils.defaultvalue import DEFAULT_TRUE
from telegram._utils.repr import build_repr_with_selected_attrs
from telegram._utils.types import DVType
from telegram.ext._utils.types import CCT, HandlerCallback
@ -95,6 +96,17 @@ class BaseHandler(Generic[UT, CCT], ABC):
self.callback: HandlerCallback[UT, CCT, RT] = callback
self.block: DVType[bool] = block
def __repr__(self) -> str:
"""Give a string representation of the handler in the form ``ClassName[callback=...]``.
As this class doesn't implement :meth:`object.__str__`, the default implementation
will be used, which is equivalent to :meth:`__repr__`.
Returns:
:obj:`str`
"""
return build_repr_with_selected_attrs(self, callback=self.callback.__qualname__)
@abstractmethod
def check_update(self, update: object) -> Optional[Union[bool, object]]:
"""

View file

@ -38,6 +38,7 @@ from typing import (
from telegram import Update
from telegram._utils.defaultvalue import DEFAULT_TRUE, DefaultValue
from telegram._utils.logging import get_logger
from telegram._utils.repr import build_repr_with_selected_attrs
from telegram._utils.types import DVType
from telegram._utils.warnings import warn
from telegram.ext._application import ApplicationHandlerStop
@ -440,6 +441,30 @@ class ConversationHandler(BaseHandler[Update, CCT]):
stacklevel=2,
)
def __repr__(self) -> str:
"""Give a string representation of the ConversationHandler in the form
``ConversationHandler[name=..., states={...}]``.
If there are more than 3 states, only the first 3 states are listed.
As this class doesn't implement :meth:`object.__str__`, the default implementation
will be used, which is equivalent to :meth:`__repr__`.
Returns:
:obj:`str`
"""
truncation_threshold = 3
states = dict(list(self.states.items())[:truncation_threshold])
states_string = str(states)
if len(self.states) > truncation_threshold:
states_string = states_string[:-1] + ", ...}"
return build_repr_with_selected_attrs(
self,
name=self.name,
states=states_string,
)
@property
def entry_points(self) -> List[BaseHandler[Update, CCT]]:
"""List[:class:`telegram.ext.BaseHandler`]: A list of :obj:`BaseHandler` objects that can

View file

@ -85,6 +85,7 @@ from telegram import (
from telegram._utils.datetime import to_timestamp
from telegram._utils.defaultvalue import DEFAULT_NONE, DefaultValue
from telegram._utils.logging import get_logger
from telegram._utils.repr import build_repr_with_selected_attrs
from telegram._utils.types import (
CorrectOptionID,
DVInput,
@ -246,6 +247,17 @@ class ExtBot(Bot, Generic[RLARGS]):
self._callback_data_cache = CallbackDataCache(bot=self, maxsize=maxsize)
def __repr__(self) -> str:
"""Give a string representation of the bot in the form ``ExtBot[token=...]``.
As this class doesn't implement :meth:`object.__str__`, the default implementation
will be used, which is equivalent to :meth:`__repr__`.
Returns:
:obj:`str`
"""
return build_repr_with_selected_attrs(self, token=self.token)
@classmethod
def _warn(
cls, message: str, category: Type[Warning] = PTBUserWarning, stacklevel: int = 0

View file

@ -31,6 +31,7 @@ try:
except ImportError:
APS_AVAILABLE = False
from telegram._utils.repr import build_repr_with_selected_attrs
from telegram._utils.types import JSONDict
from telegram._utils.warnings import warn
from telegram.ext._extbot import ExtBot
@ -97,6 +98,17 @@ class JobQueue(Generic[CCT]):
timezone=pytz.utc, executors={"default": self._executor}
)
def __repr__(self) -> str:
"""Give a string representation of the JobQueue in the form ``JobQueue[application=...]``.
As this class doesn't implement :meth:`object.__str__`, the default implementation
will be used, which is equivalent to :meth:`__repr__`.
Returns:
:obj:`str`
"""
return build_repr_with_selected_attrs(self, application=self.application)
def _tz_now(self) -> datetime.datetime:
return datetime.datetime.now(self.scheduler.timezone)
@ -766,6 +778,24 @@ class Job(Generic[CCT]):
self._job = cast("APSJob", None) # skipcq: PTC-W0052
def __repr__(self) -> str:
"""Give a string representation of the job in the form
``Job[id=..., name=..., callback=..., trigger=...]``.
As this class doesn't implement :meth:`object.__str__`, the default implementation
will be used, which is equivalent to :meth:`__repr__`.
Returns:
:obj:`str`
"""
return build_repr_with_selected_attrs(
self,
id=self.job.id,
name=self.name,
callback=self.callback.__name__,
trigger=self.job.trigger,
)
@property
def job(self) -> "APSJob":
""":class:`apscheduler.job.Job`: The APS Job this job is a wrapper for.

View file

@ -37,6 +37,7 @@ from typing import (
from telegram._utils.defaultvalue import DEFAULT_NONE
from telegram._utils.logging import get_logger
from telegram._utils.repr import build_repr_with_selected_attrs
from telegram._utils.types import ODVInput
from telegram.error import InvalidToken, RetryAfter, TelegramError, TimedOut
@ -124,6 +125,17 @@ class Updater(AsyncContextManager["Updater"]):
self.__polling_task: Optional[asyncio.Task] = None
self.__polling_cleanup_cb: Optional[Callable[[], Coroutine[Any, Any, None]]] = None
def __repr__(self) -> str:
"""Give a string representation of the updater in the form ``Updater[bot=...]``.
As this class doesn't implement :meth:`object.__str__`, the default implementation
will be used, which is equivalent to :meth:`__repr__`.
Returns:
:obj:`str`
"""
return build_repr_with_selected_attrs(self, bot=self.bot)
@property
def running(self) -> bool:
return self._running

View file

@ -201,6 +201,9 @@ class TestApplication:
assert isinstance(app.chat_data[1], dict)
assert isinstance(app.user_data[1], dict)
async def test_repr(self, app):
assert repr(app) == f"PytestApplication[bot={app.bot!r}]"
def test_job_queue(self, one_time_bot, app, recwarn):
expected_warning = (
"No `JobQueue` set up. To use `JobQueue`, you must install PTB via "

View file

@ -36,3 +36,19 @@ class TestHandler:
for attr in inst.__slots__:
assert getattr(inst, attr, "err") != "err", f"got extra slot '{attr}'"
assert len(mro_slots(inst)) == len(set(mro_slots(inst))), "duplicate slot"
def test_repr(self):
async def some_func():
return None
class SubclassHandler(BaseHandler):
__slots__ = ()
def __init__(self):
super().__init__(callback=some_func)
def check_update(self, update: object):
pass
sh = SubclassHandler()
assert repr(sh) == "SubclassHandler[callback=TestHandler.test_repr.<locals>.some_func]"

View file

@ -18,6 +18,7 @@
# along with this program. If not, see [http://www.gnu.org/licenses/].
"""Persistence of conversations is tested in test_basepersistence.py"""
import asyncio
import functools
import logging
from pathlib import Path
from warnings import filterwarnings
@ -75,6 +76,7 @@ def user2():
def raise_ahs(func):
@functools.wraps(func) # for checking __repr__
async def decorator(self, *args, **kwargs):
result = await func(self, *args, **kwargs)
if self.raise_app_handler_stop:
@ -289,6 +291,41 @@ class TestConversationHandler:
self.entry_points, states=self.states, fallbacks=[], persistent=True
)
def test_repr_no_truncation(self):
# ConversationHandler's __repr__ is not inherited from BaseHandler.
ch = ConversationHandler(
name="test_handler",
entry_points=[],
states=self.drinking_states,
fallbacks=[],
)
assert repr(ch) == (
"ConversationHandler[name=test_handler, "
"states={'a': [CommandHandler[callback=TestConversationHandler.sip]], "
"'b': [CommandHandler[callback=TestConversationHandler.swallow]], "
"'c': [CommandHandler[callback=TestConversationHandler.hold]]}]"
)
def test_repr_with_truncation(self):
from copy import copy
states = copy(self.drinking_states)
# there are exactly 3 drinking states. adding one more to make sure it's truncated
states["extra_to_be_truncated"] = [CommandHandler("foo", self.start)]
ch = ConversationHandler(
name="test_handler",
entry_points=[],
states=states,
fallbacks=[],
)
assert repr(ch) == (
"ConversationHandler[name=test_handler, "
"states={'a': [CommandHandler[callback=TestConversationHandler.sip]], "
"'b': [CommandHandler[callback=TestConversationHandler.swallow]], "
"'c': [CommandHandler[callback=TestConversationHandler.hold]], ...}]"
)
async def test_check_update_returns_non(self, app, user1):
"""checks some cases where updates should not be handled"""
conv_handler = ConversationHandler([], {}, [], per_message=True, per_chat=True)

View file

@ -85,6 +85,21 @@ class TestJobQueue:
" We recommend double checking if the passed value is correct."
)
async def test_repr(self, app):
jq = JobQueue()
jq.set_application(app)
assert repr(jq) == f"JobQueue[application={app!r}]"
when = dtm.datetime.utcnow() + dtm.timedelta(days=1)
callback = self.job_run_once
job = jq.run_once(callback, when, name="name2")
assert repr(job) == (
f"Job[id={job.job.id}, name={job.name}, callback=job_run_once, "
f"trigger=date["
f"{when.strftime('%Y-%m-%d %H:%M:%S UTC')}"
f"]]"
)
@pytest.fixture(autouse=True)
def _reset(self):
self.result = 0

View file

@ -94,6 +94,11 @@ class TestUpdater:
assert updater.bot is bot
assert updater.update_queue is queue
def test_repr(self, bot):
queue = asyncio.Queue()
updater = Updater(bot=bot, update_queue=queue)
assert repr(updater) == f"Updater[bot={updater.bot!r}]"
async def test_initialize(self, bot, monkeypatch):
async def initialize_bot(*args, **kwargs):
self.test_flag = True

View file

@ -218,6 +218,10 @@ class TestBotWithoutRequest:
with pytest.raises(InvalidToken, match="You must pass the token"):
Bot("")
async def test_repr(self):
bot = Bot(token="some_token", base_file_url="")
assert repr(bot) == "Bot[token=some_token]"
async def test_to_dict(self, bot):
to_dict_bot = bot.to_dict()