From 267d4e819113fbf3a1678e66c7b3a39a8da12ef3 Mon Sep 17 00:00:00 2001 From: Harshil <37377066+harshil21@users.noreply.github.com> Date: Sun, 3 Oct 2021 23:36:07 +0530 Subject: [PATCH] Improve Signature Inspection for Bot Methods (#2686) --- telegram/bot.py | 194 +++++++++++++++++++++++----------------------- tests/test_bot.py | 9 +++ 2 files changed, 106 insertions(+), 97 deletions(-) diff --git a/telegram/bot.py b/telegram/bot.py index 04a7365e9..7c8d013bd 100644 --- a/telegram/bot.py +++ b/telegram/bot.py @@ -35,6 +35,7 @@ from typing import ( Dict, cast, Sequence, + Any, ) try: @@ -111,22 +112,6 @@ if TYPE_CHECKING: RT = TypeVar('RT') -def log( # skipcq: PY-D0003 - func: Callable[..., RT], *args: object, **kwargs: object # pylint: disable=W0613 -) -> Callable[..., RT]: - logger = logging.getLogger(func.__module__) - - @functools.wraps(func) - def decorator(*args: object, **kwargs: object) -> RT: # pylint: disable=W0613 - logger.debug('Entering: %s', func.__name__) - result = func(*args, **kwargs) - logger.debug(result) - logger.debug('Exiting: %s', func.__name__) - return result - - return decorator - - class Bot(TelegramObject): """This object represents a Telegram Bot. @@ -203,6 +188,21 @@ class Bot(TelegramObject): private_key, password=private_key_password, backend=default_backend() ) + # 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 + logger = logging.getLogger(func.__module__) + + @functools.wraps(func) + def decorator(*args, **kwargs): # type: ignore[no-untyped-def] # pylint: disable=W0613 + logger.debug('Entering: %s', func.__name__) + result = func(*args, **kwargs) + logger.debug(result) + logger.debug('Exiting: %s', func.__name__) + return result + + return decorator + def _insert_defaults( # pylint: disable=no-self-use self, data: Dict[str, object], timeout: ODVInput[float] ) -> Optional[float]: @@ -381,7 +381,7 @@ class Bot(TelegramObject): """:obj:`str`: Bot's @username.""" return f'@{self.username}' - @log + @_log def get_me(self, timeout: ODVInput[float] = DEFAULT_NONE, api_kwargs: JSONDict = None) -> User: """A simple method for testing your bot's auth token. Requires no parameters. @@ -406,7 +406,7 @@ class Bot(TelegramObject): return self._bot # type: ignore[return-value] - @log + @_log def send_message( self, chat_id: Union[int, str], @@ -484,7 +484,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def delete_message( self, chat_id: Union[str, int], @@ -530,7 +530,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def forward_message( self, chat_id: Union[int, str], @@ -594,7 +594,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def send_photo( self, chat_id: Union[int, str], @@ -691,7 +691,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def send_audio( self, chat_id: Union[int, str], @@ -819,7 +819,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def send_document( self, chat_id: Union[int, str], @@ -935,7 +935,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def send_sticker( self, chat_id: Union[int, str], @@ -1006,7 +1006,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def send_video( self, chat_id: Union[int, str], @@ -1139,7 +1139,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def send_video_note( self, chat_id: Union[int, str], @@ -1245,7 +1245,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def send_animation( self, chat_id: Union[int, str], @@ -1369,7 +1369,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def send_voice( self, chat_id: Union[int, str], @@ -1474,7 +1474,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def send_media_group( self, chat_id: Union[int, str], @@ -1534,7 +1534,7 @@ class Bot(TelegramObject): return Message.de_list(result, self) # type: ignore - @log + @_log def send_location( self, chat_id: Union[int, str], @@ -1637,7 +1637,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def edit_message_live_location( self, chat_id: Union[str, int] = None, @@ -1726,7 +1726,7 @@ class Bot(TelegramObject): api_kwargs=api_kwargs, ) - @log + @_log def stop_message_live_location( self, chat_id: Union[str, int] = None, @@ -1776,7 +1776,7 @@ class Bot(TelegramObject): api_kwargs=api_kwargs, ) - @log + @_log def send_venue( self, chat_id: Union[int, str], @@ -1895,7 +1895,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def send_contact( self, chat_id: Union[int, str], @@ -1988,7 +1988,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def send_game( self, chat_id: Union[int, str], @@ -2048,7 +2048,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def send_chat_action( self, chat_id: Union[str, int], @@ -2165,7 +2165,7 @@ class Bot(TelegramObject): res.input_message_content.disable_web_page_preview ) - @log + @_log def answer_inline_query( self, inline_query_id: str, @@ -2268,7 +2268,7 @@ class Bot(TelegramObject): api_kwargs=api_kwargs, ) - @log + @_log def get_user_profile_photos( self, user_id: Union[str, int], @@ -2309,7 +2309,7 @@ class Bot(TelegramObject): return UserProfilePhotos.de_json(result, self) # type: ignore[return-value, arg-type] - @log + @_log def get_file( self, file_id: Union[ @@ -2369,7 +2369,7 @@ class Bot(TelegramObject): return File.de_json(result, self) # type: ignore[return-value, arg-type] - @log + @_log def ban_chat_member( self, chat_id: Union[str, int], @@ -2428,7 +2428,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def ban_chat_sender_chat( self, chat_id: Union[str, int], @@ -2467,7 +2467,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def unban_chat_member( self, chat_id: Union[str, int], @@ -2511,7 +2511,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def unban_chat_sender_chat( self, chat_id: Union[str, int], @@ -2548,7 +2548,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def answer_callback_query( self, callback_query_id: str, @@ -2611,7 +2611,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def edit_message_text( self, text: str, @@ -2683,7 +2683,7 @@ class Bot(TelegramObject): api_kwargs=api_kwargs, ) - @log + @_log def edit_message_caption( self, chat_id: Union[str, int] = None, @@ -2758,7 +2758,7 @@ class Bot(TelegramObject): api_kwargs=api_kwargs, ) - @log + @_log def edit_message_media( self, media: 'InputMedia', @@ -2824,7 +2824,7 @@ class Bot(TelegramObject): api_kwargs=api_kwargs, ) - @log + @_log def edit_message_reply_markup( self, chat_id: Union[str, int] = None, @@ -2885,7 +2885,7 @@ class Bot(TelegramObject): api_kwargs=api_kwargs, ) - @log + @_log def get_updates( self, offset: int = None, @@ -2969,7 +2969,7 @@ class Bot(TelegramObject): return Update.de_list(result, self) # type: ignore[return-value] - @log + @_log def set_webhook( self, url: str, @@ -3060,7 +3060,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def delete_webhook( self, timeout: ODVInput[float] = DEFAULT_NONE, @@ -3096,7 +3096,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def leave_chat( self, chat_id: Union[str, int], @@ -3127,7 +3127,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def get_chat( self, chat_id: Union[str, int], @@ -3160,7 +3160,7 @@ class Bot(TelegramObject): return Chat.de_json(result, self) # type: ignore[return-value, arg-type] - @log + @_log def get_chat_administrators( self, chat_id: Union[str, int], @@ -3195,7 +3195,7 @@ class Bot(TelegramObject): return ChatMember.de_list(result, self) # type: ignore - @log + @_log def get_chat_member_count( self, chat_id: Union[str, int], @@ -3228,7 +3228,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def get_chat_member( self, chat_id: Union[str, int], @@ -3261,7 +3261,7 @@ class Bot(TelegramObject): return ChatMember.de_json(result, self) # type: ignore[return-value, arg-type] - @log + @_log def set_chat_sticker_set( self, chat_id: Union[str, int], @@ -3294,7 +3294,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def delete_chat_sticker_set( self, chat_id: Union[str, int], @@ -3347,7 +3347,7 @@ class Bot(TelegramObject): return WebhookInfo.de_json(result, self) # type: ignore[return-value, arg-type] - @log + @_log def set_game_score( self, user_id: Union[int, str], @@ -3411,7 +3411,7 @@ class Bot(TelegramObject): api_kwargs=api_kwargs, ) - @log + @_log def get_game_high_scores( self, user_id: Union[int, str], @@ -3464,7 +3464,7 @@ class Bot(TelegramObject): return GameHighScore.de_list(result, self) # type: ignore - @log + @_log def send_invoice( self, chat_id: Union[int, str], @@ -3650,7 +3650,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def answer_shipping_query( # pylint: disable=C0103 self, shipping_query_id: str, @@ -3719,7 +3719,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def answer_pre_checkout_query( # pylint: disable=C0103 self, pre_checkout_query_id: str, @@ -3778,7 +3778,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def restrict_chat_member( self, chat_id: Union[str, int], @@ -3835,7 +3835,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def promote_chat_member( self, chat_id: Union[str, int], @@ -3937,7 +3937,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def set_chat_permissions( self, chat_id: Union[str, int], @@ -3973,7 +3973,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def set_chat_administrator_custom_title( self, chat_id: Union[int, str], @@ -4013,7 +4013,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def export_chat_invite_link( self, chat_id: Union[str, int], @@ -4054,7 +4054,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def create_chat_invite_link( self, chat_id: Union[str, int], @@ -4127,7 +4127,7 @@ class Bot(TelegramObject): return ChatInviteLink.de_json(result, self) # type: ignore[return-value, arg-type] - @log + @_log def edit_chat_invite_link( self, chat_id: Union[str, int], @@ -4205,7 +4205,7 @@ class Bot(TelegramObject): return ChatInviteLink.de_json(result, self) # type: ignore[return-value, arg-type] - @log + @_log def revoke_chat_invite_link( self, chat_id: Union[str, int], @@ -4243,7 +4243,7 @@ class Bot(TelegramObject): return ChatInviteLink.de_json(result, self) # type: ignore[return-value, arg-type] - @log + @_log def approve_chat_join_request( self, chat_id: Union[str, int], @@ -4280,7 +4280,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def decline_chat_join_request( self, chat_id: Union[str, int], @@ -4317,7 +4317,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def set_chat_photo( self, chat_id: Union[str, int], @@ -4356,7 +4356,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def delete_chat_photo( self, chat_id: Union[str, int], @@ -4390,7 +4390,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def set_chat_title( self, chat_id: Union[str, int], @@ -4426,7 +4426,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def set_chat_description( self, chat_id: Union[str, int], @@ -4465,7 +4465,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def pin_chat_message( self, chat_id: Union[str, int], @@ -4510,7 +4510,7 @@ class Bot(TelegramObject): 'pinChatMessage', data, timeout=timeout, api_kwargs=api_kwargs ) - @log + @_log def unpin_chat_message( self, chat_id: Union[str, int], @@ -4551,7 +4551,7 @@ class Bot(TelegramObject): 'unpinChatMessage', data, timeout=timeout, api_kwargs=api_kwargs ) - @log + @_log def unpin_all_chat_messages( self, chat_id: Union[str, int], @@ -4586,7 +4586,7 @@ class Bot(TelegramObject): 'unpinAllChatMessages', data, timeout=timeout, api_kwargs=api_kwargs ) - @log + @_log def get_sticker_set( self, name: str, @@ -4616,7 +4616,7 @@ class Bot(TelegramObject): return StickerSet.de_json(result, self) # type: ignore[return-value, arg-type] - @log + @_log def upload_sticker_file( self, user_id: Union[str, int], @@ -4660,7 +4660,7 @@ class Bot(TelegramObject): return File.de_json(result, self) # type: ignore[return-value, arg-type] - @log + @_log def create_new_sticker_set( self, user_id: Union[str, int], @@ -4759,7 +4759,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def add_sticker_to_set( self, user_id: Union[str, int], @@ -4848,7 +4848,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def set_sticker_position_in_set( self, sticker: str, @@ -4882,7 +4882,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def delete_sticker_from_set( self, sticker: str, @@ -4912,7 +4912,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def set_sticker_set_thumb( self, name: str, @@ -4968,7 +4968,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def set_passport_data_errors( self, user_id: Union[str, int], @@ -5009,7 +5009,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def send_poll( self, chat_id: Union[int, str], @@ -5133,7 +5133,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def stop_poll( self, chat_id: Union[int, str], @@ -5178,7 +5178,7 @@ class Bot(TelegramObject): return Poll.de_json(result, self) # type: ignore[return-value, arg-type] - @log + @_log def send_dice( self, chat_id: Union[int, str], @@ -5248,7 +5248,7 @@ class Bot(TelegramObject): protect_content=protect_content, ) - @log + @_log def get_my_commands( self, timeout: ODVInput[float] = DEFAULT_NONE, @@ -5296,7 +5296,7 @@ class Bot(TelegramObject): return BotCommand.de_list(result, self) # type: ignore[return-value,arg-type] - @log + @_log def set_my_commands( self, commands: List[Union[BotCommand, Tuple[str, str]]], @@ -5352,7 +5352,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def delete_my_commands( self, scope: BotCommandScope = None, @@ -5399,7 +5399,7 @@ class Bot(TelegramObject): return result # type: ignore[return-value] - @log + @_log def log_out(self, timeout: ODVInput[float] = DEFAULT_NONE) -> bool: """ Use this method to log out from the cloud Bot API server before launching the bot locally. @@ -5422,7 +5422,7 @@ class Bot(TelegramObject): """ return self._post('logOut', timeout=timeout) # type: ignore[return-value] - @log + @_log def close(self, timeout: ODVInput[float] = DEFAULT_NONE) -> bool: """ Use this method to close the bot instance before moving it from one local server to @@ -5444,7 +5444,7 @@ class Bot(TelegramObject): """ return self._post('close', timeout=timeout) # type: ignore[return-value] - @log + @_log def copy_message( self, chat_id: Union[int, str], diff --git a/tests/test_bot.py b/tests/test_bot.py index 21de95f7e..88651d687 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -17,6 +17,7 @@ # You should have received a copy of the GNU Lesser Public License # along with this program. If not, see [http://www.gnu.org/licenses/]. import inspect +import logging import time import datetime as dtm from collections import defaultdict @@ -165,6 +166,13 @@ class TestBot: with pytest.raises(InvalidToken, match='Invalid token'): Bot(token) + def test_log_decorator(self, bot, caplog): + with caplog.at_level(logging.DEBUG): + bot.get_me() + assert len(caplog.records) == 3 + assert caplog.records[0].getMessage().startswith('Entering: get_me') + assert caplog.records[-1].getMessage().startswith('Exiting: get_me') + @pytest.mark.parametrize( 'acd_in,maxsize,acd', [(True, 1024, True), (False, 1024, False), (0, 0, True), (None, None, True)], @@ -243,6 +251,7 @@ class TestBot: 'de_list', 'to_dict', 'to_json', + 'log', 'parse_data', 'get_updates', 'getUpdates',