OpenSource/python-telegram-bot
OpenSource/python-telegram-bot
1
0
Fork 0
mirror of https://github.com/python-telegram-bot/python-telegram-bot.git synced 2024-12-22 06:25:12 +01:00
Code Issues Projects Releases Packages Wiki Activity

Verify Type Hints for Bot Method & Telegram Class Parameters (#3868)

Browse source
This commit is contained in:
Harshil 2023-09-15 23:33:42 +04:00 committed by GitHub
parent 04b44f4595
commit 39abf838fa
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 495 additions and 139 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

72
telegram/_bot.py
View file

@ -69,7 +69,6 @@ from telegram._files.contact import Contact
from telegram._files.document import Document
from telegram._files.file import File
from telegram._files.inputmedia import InputMedia
from telegram._files.inputsticker import InputSticker
from telegram._files.location import Location
from telegram._files.photosize import PhotoSize
from telegram._files.sticker import MaskPosition, Sticker, StickerSet
@ -79,13 +78,10 @@ from telegram._files.videonote import VideoNote
from telegram._files.voice import Voice
from telegram._forumtopic import ForumTopic
from telegram._games.gamehighscore import GameHighScore
from telegram._inline.inlinekeyboardmarkup import InlineKeyboardMarkup
from telegram._inline.inlinequeryresultsbutton import InlineQueryResultsButton
from telegram._menubutton import MenuButton
from telegram._message import Message
from telegram._messageid import MessageId
from telegram._passport.passportelementerrors import PassportElementError
from telegram._payment.shippingoption import ShippingOption
from telegram._poll import Poll
from telegram._sentwebappmessage import SentWebAppMessage
from telegram._telegramobject import TelegramObject
@ -115,14 +111,18 @@ from telegram.warnings import PTBUserWarning
if TYPE_CHECKING:
from telegram import (
InlineKeyboardMarkup,
InlineQueryResult,
InputFile,
InputMediaAudio,
InputMediaDocument,
InputMediaPhoto,
InputMediaVideo,
InputSticker,
LabeledPrice,
MessageEntity,
PassportElementError,
ShippingOption,
)
BT = TypeVar("BT", bound="Bot")
@ -2154,7 +2154,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
inline_message_id: Optional[str] = None,
latitude: Optional[float] = None,
longitude: Optional[float] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
horizontal_accuracy: Optional[float] = None,
heading: Optional[int] = None,
proximity_alert_radius: Optional[int] = None,
@ -2247,7 +2247,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
chat_id: Optional[Union[str, int]] = None,
message_id: Optional[int] = None,
inline_message_id: Optional[str] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -2521,11 +2521,11 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
@_log
async def send_game(
self,
chat_id: Union[int, str],
chat_id: int,
game_short_name: str,
disable_notification: DVInput[bool] = DEFAULT_NONE,
reply_to_message_id: Optional[int] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
allow_sending_without_reply: ODVInput[bool] = DEFAULT_NONE,
protect_content: ODVInput[bool] = DEFAULT_NONE,
message_thread_id: Optional[int] = None,
@ -2539,7 +2539,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
"""Use this method to send a game.
Args:
chat_id (:obj:`int` | :obj:`str`): Unique identifier for the target chat.
chat_id (:obj:`int`): Unique identifier for the target chat.
game_short_name (:obj:`str`): Short name of the game, serves as the unique identifier
for the game. Set up your games via `@BotFather <https://t.me/BotFather>`_.
disable_notification (:obj:`bool`, optional): |disable_notification|
@ -2826,7 +2826,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
@_log
async def get_user_profile_photos(
self,
user_id: Union[str, int],
user_id: int,
offset: Optional[int] = None,
limit: Optional[int] = None,
*,
@ -2938,7 +2938,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
async def ban_chat_member(
self,
chat_id: Union[str, int],
user_id: Union[str, int],
user_id: int,
until_date: Optional[Union[int, datetime]] = None,
revoke_messages: Optional[bool] = None,
*,
@ -3046,7 +3046,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
async def unban_chat_member(
self,
chat_id: Union[str, int],
user_id: Union[str, int],
user_id: int,
only_if_banned: Optional[bool] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -3203,7 +3203,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
inline_message_id: Optional[str] = None,
parse_mode: ODVInput[str] = DEFAULT_NONE,
disable_web_page_preview: ODVInput[bool] = DEFAULT_NONE,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
entities: Optional[Sequence["MessageEntity"]] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -3279,7 +3279,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
message_id: Optional[int] = None,
inline_message_id: Optional[str] = None,
caption: Optional[str] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
parse_mode: ODVInput[str] = DEFAULT_NONE,
caption_entities: Optional[Sequence["MessageEntity"]] = None,
*,
@ -3349,7 +3349,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
chat_id: Optional[Union[str, int]] = None,
message_id: Optional[int] = None,
inline_message_id: Optional[str] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -3876,7 +3876,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
async def get_chat_member(
self,
chat_id: Union[str, int],
user_id: Union[str, int],
user_id: int,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -4011,9 +4011,9 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
@_log
async def set_game_score(
self,
user_id: Union[int, str],
user_id: int,
score: int,
chat_id: Optional[Union[str, int]] = None,
chat_id: Optional[int] = None,
message_id: Optional[int] = None,
inline_message_id: Optional[str] = None,
force: Optional[bool] = None,
@ -4037,7 +4037,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
decrease. This can be useful when fixing mistakes or banning cheaters.
disable_edit_message (:obj:`bool`, optional): Pass :obj:`True`, if the game message
should not be automatically edited to include the current scoreboard.
chat_id (:obj:`int` | :obj:`str`, optional): Required if :paramref:`inline_message_id`
chat_id (:obj:`int`, optional): Required if :paramref:`inline_message_id`
is not specified. Unique identifier for the target chat.
message_id (:obj:`int`, optional): Required if :paramref:`inline_message_id` is not
specified. Identifier of the sent message.
@ -4076,8 +4076,8 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
@_log
async def get_game_high_scores(
self,
user_id: Union[int, str],
chat_id: Optional[Union[str, int]] = None,
user_id: int,
chat_id: Optional[int] = None,
message_id: Optional[int] = None,
inline_message_id: Optional[str] = None,
*,
@ -4101,7 +4101,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
Args:
user_id (:obj:`int`): Target user id.
chat_id (:obj:`int` | :obj:`str`, optional): Required if :paramref:`inline_message_id`
chat_id (:obj:`int`, optional): Required if :paramref:`inline_message_id`
is not specified. Unique identifier for the target chat.
message_id (:obj:`int`, optional): Required if :paramref:`inline_message_id` is not
specified. Identifier of the sent message.
@ -4156,7 +4156,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
is_flexible: Optional[bool] = None,
disable_notification: DVInput[bool] = DEFAULT_NONE,
reply_to_message_id: Optional[int] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
provider_data: Optional[Union[str, object]] = None,
send_phone_number_to_provider: Optional[bool] = None,
send_email_to_provider: Optional[bool] = None,
@ -4321,7 +4321,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
self,
shipping_query_id: str,
ok: bool,
shipping_options: Optional[Sequence[ShippingOption]] = None,
shipping_options: Optional[Sequence["ShippingOption"]] = None,
error_message: Optional[str] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -4483,7 +4483,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
async def restrict_chat_member(
self,
chat_id: Union[str, int],
user_id: Union[str, int],
user_id: int,
permissions: ChatPermissions,
until_date: Optional[Union[int, datetime]] = None,
use_independent_chat_permissions: Optional[bool] = None,
@ -4557,7 +4557,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
async def promote_chat_member(
self,
chat_id: Union[str, int],
user_id: Union[str, int],
user_id: int,
can_change_info: Optional[bool] = None,
can_post_messages: Optional[bool] = None,
can_edit_messages: Optional[bool] = None,
@ -4723,7 +4723,7 @@ class Bot(TelegramObject, AsyncContextManager["Bot"]):
async def set_chat_administrator_custom_title(
self,
chat_id: Union[int, str],
user_id: Union[int, str],
user_id: int,
custom_title: str,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -5478,7 +5478,7 @@ CUSTOM_EMOJI_IDENTIFIER_LIMIT` custom emoji identifiers can be specified.
@_log
async def upload_sticker_file(
self,
user_id: Union[str, int],
user_id: int,
sticker: Optional[FileInput],
sticker_format: Optional[str],
*,
@ -5538,9 +5538,9 @@ CUSTOM_EMOJI_IDENTIFIER_LIMIT` custom emoji identifiers can be specified.
@_log
async def add_sticker_to_set(
self,
user_id: Union[str, int],
user_id: int,
name: str,
sticker: Optional[InputSticker],
sticker: Optional["InputSticker"],
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = 20,
@ -5636,10 +5636,10 @@ CUSTOM_EMOJI_IDENTIFIER_LIMIT` custom emoji identifiers can be specified.
@_log
async def create_new_sticker_set(
self,
user_id: Union[str, int],
user_id: int,
name: str,
title: str,
stickers: Optional[Sequence[InputSticker]],
stickers: Optional[Sequence["InputSticker"]],
sticker_format: Optional[str],
sticker_type: Optional[str] = None,
needs_repainting: Optional[bool] = None,
@ -5807,7 +5807,7 @@ CUSTOM_EMOJI_IDENTIFIER_LIMIT` custom emoji identifiers can be specified.
async def set_sticker_set_thumbnail(
self,
name: str,
user_id: Union[str, int],
user_id: int,
thumbnail: Optional[FileInput] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -6079,8 +6079,8 @@ CUSTOM_EMOJI_IDENTIFIER_LIMIT` custom emoji identifiers can be specified.
@_log
async def set_passport_data_errors(
self,
user_id: Union[str, int],
errors: Sequence[PassportElementError],
user_id: int,
errors: Sequence["PassportElementError"],
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -6262,7 +6262,7 @@ CUSTOM_EMOJI_IDENTIFIER_LIMIT` custom emoji identifiers can be specified.
self,
chat_id: Union[int, str],
message_id: int,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,

4
telegram/_callbackquery.py
View file

@ -531,7 +531,7 @@ class CallbackQuery(TelegramObject):
async def set_game_score(
self,
user_id: Union[int, str],
user_id: int,
score: int,
force: Optional[bool] = None,
disable_edit_message: Optional[bool] = None,
@ -589,7 +589,7 @@ class CallbackQuery(TelegramObject):
async def get_game_high_scores(
self,
user_id: Union[int, str],
user_id: int,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,

10
telegram/_chat.py
View file

@ -677,7 +677,7 @@ class Chat(TelegramObject):
async def get_member(
self,
user_id: Union[str, int],
user_id: int,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -707,7 +707,7 @@ class Chat(TelegramObject):
async def ban_member(
self,
user_id: Union[str, int],
user_id: int,
revoke_messages: Optional[bool] = None,
until_date: Optional[Union[int, datetime]] = None,
*,
@ -877,7 +877,7 @@ class Chat(TelegramObject):
async def unban_member(
self,
user_id: Union[str, int],
user_id: int,
only_if_banned: Optional[bool] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -909,7 +909,7 @@ class Chat(TelegramObject):
async def promote_member(
self,
user_id: Union[str, int],
user_id: int,
can_change_info: Optional[bool] = None,
can_post_messages: Optional[bool] = None,
can_edit_messages: Optional[bool] = None,
@ -970,7 +970,7 @@ class Chat(TelegramObject):
async def restrict_member(
self,
user_id: Union[str, int],
user_id: int,
permissions: ChatPermissions,
until_date: Optional[Union[int, datetime]] = None,
use_independent_chat_permissions: Optional[bool] = None,

4
telegram/_loginurl.py
View file

@ -86,7 +86,7 @@ class LoginUrl(TelegramObject):
def __init__(
self,
url: str,
forward_text: Optional[bool] = None,
forward_text: Optional[str] = None,
bot_username: Optional[str] = None,
request_write_access: Optional[bool] = None,
*,
@ -96,7 +96,7 @@ class LoginUrl(TelegramObject):
# Required
self.url: str = url
# Optional
self.forward_text: Optional[bool] = forward_text
self.forward_text: Optional[str] = forward_text
self.bot_username: Optional[str] = bot_username
self.request_write_access: Optional[bool] = request_write_access

16
telegram/_message.py
View file

@ -2504,7 +2504,7 @@ class Message(TelegramObject):
text: str,
parse_mode: ODVInput[str] = DEFAULT_NONE,
disable_web_page_preview: ODVInput[bool] = DEFAULT_NONE,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
entities: Optional[Sequence["MessageEntity"]] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -2550,7 +2550,7 @@ class Message(TelegramObject):
async def edit_caption(
self,
caption: Optional[str] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
parse_mode: ODVInput[str] = DEFAULT_NONE,
caption_entities: Optional[Sequence["MessageEntity"]] = None,
*,
@ -2597,7 +2597,7 @@ class Message(TelegramObject):
async def edit_media(
self,
media: "InputMedia",
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -2681,7 +2681,7 @@ class Message(TelegramObject):
self,
latitude: Optional[float] = None,
longitude: Optional[float] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
horizontal_accuracy: Optional[float] = None,
heading: Optional[int] = None,
proximity_alert_radius: Optional[int] = None,
@ -2731,7 +2731,7 @@ class Message(TelegramObject):
async def stop_live_location(
self,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -2771,7 +2771,7 @@ class Message(TelegramObject):
async def set_game_score(
self,
user_id: Union[int, str],
user_id: int,
score: int,
force: Optional[bool] = None,
disable_edit_message: Optional[bool] = None,
@ -2816,7 +2816,7 @@ class Message(TelegramObject):
async def get_game_high_scores(
self,
user_id: Union[int, str],
user_id: int,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -2886,7 +2886,7 @@ class Message(TelegramObject):
async def stop_poll(
self,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,

10
telegram/_passport/encryptedpassportelement.py
View file

@ -18,7 +18,7 @@
# along with this program. If not, see [http://www.gnu.org/licenses/].
"""This module contains an object that represents a Telegram EncryptedPassportElement."""
from base64 import b64decode
from typing import TYPE_CHECKING, Optional, Sequence, Tuple
from typing import TYPE_CHECKING, Optional, Sequence, Tuple, Union
from telegram._passport.credentials import decrypt_json
from telegram._passport.data import IdDocumentData, PersonalDetails, ResidentialAddress
@ -54,7 +54,7 @@ class EncryptedPassportElement(TelegramObject):
data (:class:`telegram.PersonalDetails` | :class:`telegram.IdDocumentData` | \
:class:`telegram.ResidentialAddress` | :obj:`str`, optional):
Decrypted or encrypted data, available for "personal_details", "passport",
"driver_license", "identity_card", "identity_passport" and "address" types.
"driver_license", "identity_card", "internal_passport" and "address" types.
phone_number (:obj:`str`, optional): User's verified phone number, available only for
"phone_number" type.
email (:obj:`str`, optional): User's verified email address, available only for "email"
@ -96,7 +96,7 @@ class EncryptedPassportElement(TelegramObject):
data (:class:`telegram.PersonalDetails` | :class:`telegram.IdDocumentData` | \
:class:`telegram.ResidentialAddress` | :obj:`str`):
Optional. Decrypted or encrypted data, available for "personal_details", "passport",
"driver_license", "identity_card", "identity_passport" and "address" types.
"driver_license", "identity_card", "internal_passport" and "address" types.
phone_number (:obj:`str`): Optional. User's verified phone number, available only for
"phone_number" type.
email (:obj:`str`): Optional. User's verified email address, available only for "email"
@ -151,7 +151,7 @@ class EncryptedPassportElement(TelegramObject):
self,
type: str, # pylint: disable=redefined-builtin
hash: str, # pylint: disable=redefined-builtin
data: Optional[PersonalDetails] = None,
data: Optional[Union[PersonalDetails, IdDocumentData, ResidentialAddress]] = None,
phone_number: Optional[str] = None,
email: Optional[str] = None,
files: Optional[Sequence[PassportFile]] = None,
@ -168,7 +168,7 @@ class EncryptedPassportElement(TelegramObject):
# Required
self.type: str = type
# Optionals
self.data: Optional[PersonalDetails] = data
self.data: Optional[Union[PersonalDetails, IdDocumentData, ResidentialAddress]] = data
self.phone_number: Optional[str] = phone_number
self.email: Optional[str] = email
self.files: Tuple[PassportFile, ...] = parse_sequence_arg(files)

71
telegram/_passport/passportelementerrors.py
View file

@ -19,10 +19,12 @@
# pylint: disable=redefined-builtin
"""This module contains the classes that represent Telegram PassportElementError."""
from typing import Optional
from typing import List, Optional
from telegram._telegramobject import TelegramObject
from telegram._utils.types import JSONDict
from telegram._utils.warnings import warn
from telegram.warnings import PTBDeprecationWarning
class PassportElementError(TelegramObject):
@ -173,23 +175,48 @@ class PassportElementErrorFiles(PassportElementError):
type (:obj:`str`): The section of the user's Telegram Passport which has the issue, one of
``"utility_bill"``, ``"bank_statement"``, ``"rental_agreement"``,
``"passport_registration"``, ``"temporary_registration"``.
file_hashes (List[:obj:`str`]): List of base64-encoded file hashes.
message (:obj:`str`): Error message.
"""
__slots__ = ("file_hashes",)
__slots__ = ("_file_hashes",)
def __init__(
self, type: str, file_hashes: str, message: str, *, api_kwargs: Optional[JSONDict] = None
self,
type: str,
file_hashes: List[str],
message: str,
*,
api_kwargs: Optional[JSONDict] = None,
):
# Required
super().__init__("files", type, message, api_kwargs=api_kwargs)
with self._unfrozen():
self.file_hashes: str = file_hashes
self._file_hashes: List[str] = file_hashes
self._id_attrs = (self.source, self.type, self.message, *tuple(file_hashes))
def to_dict(self, recursive: bool = True) -> JSONDict:
"""See :meth:`telegram.TelegramObject.to_dict` for details."""
data = super().to_dict(recursive)
data["file_hashes"] = self._file_hashes
return data
@property
def file_hashes(self) -> List[str]:
"""List of base64-encoded file hashes.
.. deprecated:: NEXT.VERSION
This attribute will return a tuple instead of a list in future major versions.
"""
warn(
"The attribute `file_hashes` will return a tuple instead of a list in future major"
" versions.",
PTBDeprecationWarning,
stacklevel=2,
)
return self._file_hashes
class PassportElementErrorFrontSide(PassportElementError):
"""
@ -365,23 +392,49 @@ class PassportElementErrorTranslationFiles(PassportElementError):
one of ``"passport"``, ``"driver_license"``, ``"identity_card"``,
``"internal_passport"``, ``"utility_bill"``, ``"bank_statement"``,
``"rental_agreement"``, ``"passport_registration"``, ``"temporary_registration"``.
file_hashes (List[:obj:`str`]): List of base64-encoded file hashes.
message (:obj:`str`): Error message.
"""
__slots__ = ("file_hashes",)
__slots__ = ("_file_hashes",)
def __init__(
self, type: str, file_hashes: str, message: str, *, api_kwargs: Optional[JSONDict] = None
self,
type: str,
file_hashes: List[str],
message: str,
*,
api_kwargs: Optional[JSONDict] = None,
):
# Required
super().__init__("translation_files", type, message, api_kwargs=api_kwargs)
with self._unfrozen():
self.file_hashes: str = file_hashes
self._file_hashes: List[str] = file_hashes
self._id_attrs = (self.source, self.type, self.message, *tuple(file_hashes))
def to_dict(self, recursive: bool = True) -> JSONDict:
"""See :meth:`telegram.TelegramObject.to_dict` for details."""
data = super().to_dict(recursive)
data["file_hashes"] = self._file_hashes
return data
@property
def file_hashes(self) -> List[str]:
"""List of base64-encoded file hashes.
.. deprecated:: NEXT.VERSION
This attribute will return a tuple instead of a list in future major versions.
"""
warn(
"The attribute `file_hashes` will return a tuple instead of a list in future major"
" versions. See the stability policy:"
" https://docs.python-telegram-bot.org/en/stable/stability_policy.html",
PTBDeprecationWarning,
stacklevel=2,
)
return self._file_hashes
class PassportElementErrorUnspecified(PassportElementError):
"""

34
telegram/_passport/passportfile.py
View file

@ -23,6 +23,8 @@ from typing import TYPE_CHECKING, List, Optional, Tuple
from telegram._telegramobject import TelegramObject
from telegram._utils.defaultvalue import DEFAULT_NONE
from telegram._utils.types import JSONDict, ODVInput
from telegram._utils.warnings import warn
from telegram.warnings import PTBDeprecationWarning
if TYPE_CHECKING:
from telegram import Bot, File, FileCredentials
@ -45,6 +47,10 @@ class PassportFile(TelegramObject):
file_size (:obj:`int`): File size in bytes.
file_date (:obj:`int`): Unix time when the file was uploaded.
.. deprecated:: NEXT.VERSION
This argument will only accept a datetime instead of an integer in future
major versions.
Attributes:
file_id (:obj:`str`): Identifier for this file, which can be used to download
or reuse the file.
@ -52,13 +58,10 @@ class PassportFile(TelegramObject):
is supposed to be the same over time and for different bots.
Can't be used to download or reuse the file.
file_size (:obj:`int`): File size in bytes.
file_date (:obj:`int`): Unix time when the file was uploaded.
"""
__slots__ = (
"file_date",
"_file_date",
"file_id",
"file_size",
"_credentials",
@ -81,7 +84,7 @@ class PassportFile(TelegramObject):
self.file_id: str = file_id
self.file_unique_id: str = file_unique_id
self.file_size: int = file_size
self.file_date: int = file_date
self._file_date: int = file_date
# Optionals
self._credentials: Optional[FileCredentials] = credentials
@ -90,6 +93,27 @@ class PassportFile(TelegramObject):
self._freeze()
def to_dict(self, recursive: bool = True) -> JSONDict:
"""See :meth:`telegram.TelegramObject.to_dict` for details."""
data = super().to_dict(recursive)
data["file_date"] = self._file_date
return data
@property
def file_date(self) -> int:
""":obj:`int`: Unix time when the file was uploaded.
.. deprecated:: NEXT.VERSION
This attribute will return a datetime instead of a integer in future major versions.
"""
warn(
"The attribute `file_date` will return a datetime instead of an integer in future"
" major versions.",
PTBDeprecationWarning,
stacklevel=2,
)
return self._file_date
@classmethod
def de_json_decrypted(
cls, data: Optional[JSONDict], bot: "Bot", credentials: "FileCredentials"

4
telegram/_payment/orderinfo.py
View file

@ -56,7 +56,7 @@ class OrderInfo(TelegramObject):
name: Optional[str] = None,
phone_number: Optional[str] = None,
email: Optional[str] = None,
shipping_address: Optional[str] = None,
shipping_address: Optional[ShippingAddress] = None,
*,
api_kwargs: Optional[JSONDict] = None,
):
@ -64,7 +64,7 @@ class OrderInfo(TelegramObject):
self.name: Optional[str] = name
self.phone_number: Optional[str] = phone_number
self.email: Optional[str] = email
self.shipping_address: Optional[str] = shipping_address
self.shipping_address: Optional[ShippingAddress] = shipping_address
self._id_attrs = (self.name, self.phone_number, self.email, self.shipping_address)

4
telegram/_payment/shippingquery.py
View file

@ -21,7 +21,6 @@
from typing import TYPE_CHECKING, Optional, Sequence
from telegram._payment.shippingaddress import ShippingAddress
from telegram._payment.shippingoption import ShippingOption
from telegram._telegramobject import TelegramObject
from telegram._user import User
from telegram._utils.defaultvalue import DEFAULT_NONE
@ -29,6 +28,7 @@ from telegram._utils.types import JSONDict, ODVInput
if TYPE_CHECKING:
from telegram import Bot
from telegram._payment.shippingoption import ShippingOption
class ShippingQuery(TelegramObject):
@ -92,7 +92,7 @@ class ShippingQuery(TelegramObject):
async def answer(
self,
ok: bool,
shipping_options: Optional[Sequence[ShippingOption]] = None,
shipping_options: Optional[Sequence["ShippingOption"]] = None,
error_message: Optional[str] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,

64
telegram/ext/_extbot.py
View file

@ -63,17 +63,14 @@ from telegram import (
InlineKeyboardMarkup,
InlineQueryResultsButton,
InputMedia,
InputSticker,
Location,
MaskPosition,
MenuButton,
Message,
MessageId,
PassportElementError,
PhotoSize,
Poll,
SentWebAppMessage,
ShippingOption,
Sticker,
StickerSet,
Update,
@ -108,8 +105,11 @@ if TYPE_CHECKING:
InputMediaDocument,
InputMediaPhoto,
InputMediaVideo,
InputSticker,
LabeledPrice,
MessageEntity,
PassportElementError,
ShippingOption,
)
from telegram.ext import BaseRateLimiter, Defaults
@ -645,7 +645,7 @@ class ExtBot(Bot, Generic[RLARGS]):
self,
chat_id: Union[int, str],
message_id: int,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -733,9 +733,9 @@ class ExtBot(Bot, Generic[RLARGS]):
async def add_sticker_to_set(
self,
user_id: Union[str, int],
user_id: int,
name: str,
sticker: Optional[InputSticker],
sticker: Optional["InputSticker"],
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = 20,
@ -845,7 +845,7 @@ class ExtBot(Bot, Generic[RLARGS]):
self,
shipping_query_id: str,
ok: bool,
shipping_options: Optional[Sequence[ShippingOption]] = None,
shipping_options: Optional[Sequence["ShippingOption"]] = None,
error_message: Optional[str] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -914,7 +914,7 @@ class ExtBot(Bot, Generic[RLARGS]):
async def ban_chat_member(
self,
chat_id: Union[str, int],
user_id: Union[str, int],
user_id: int,
until_date: Optional[Union[int, datetime]] = None,
revoke_messages: Optional[bool] = None,
*,
@ -1047,10 +1047,10 @@ class ExtBot(Bot, Generic[RLARGS]):
async def create_new_sticker_set(
self,
user_id: Union[str, int],
user_id: int,
name: str,
title: str,
stickers: Optional[Sequence[InputSticker]],
stickers: Optional[Sequence["InputSticker"]],
sticker_format: Optional[str],
sticker_type: Optional[str] = None,
needs_repainting: Optional[bool] = None,
@ -1329,7 +1329,7 @@ class ExtBot(Bot, Generic[RLARGS]):
message_id: Optional[int] = None,
inline_message_id: Optional[str] = None,
caption: Optional[str] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
parse_mode: ODVInput[str] = DEFAULT_NONE,
caption_entities: Optional[Sequence["MessageEntity"]] = None,
*,
@ -1362,7 +1362,7 @@ class ExtBot(Bot, Generic[RLARGS]):
inline_message_id: Optional[str] = None,
latitude: Optional[float] = None,
longitude: Optional[float] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
horizontal_accuracy: Optional[float] = None,
heading: Optional[int] = None,
proximity_alert_radius: Optional[int] = None,
@ -1399,7 +1399,7 @@ class ExtBot(Bot, Generic[RLARGS]):
chat_id: Optional[Union[str, int]] = None,
message_id: Optional[int] = None,
inline_message_id: Optional[str] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -1455,7 +1455,7 @@ class ExtBot(Bot, Generic[RLARGS]):
inline_message_id: Optional[str] = None,
parse_mode: ODVInput[str] = DEFAULT_NONE,
disable_web_page_preview: ODVInput[bool] = DEFAULT_NONE,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
entities: Optional[Sequence["MessageEntity"]] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -1554,7 +1554,7 @@ class ExtBot(Bot, Generic[RLARGS]):
async def get_chat_member(
self,
chat_id: Union[str, int],
user_id: Union[str, int],
user_id: int,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -1655,8 +1655,8 @@ class ExtBot(Bot, Generic[RLARGS]):
async def get_game_high_scores(
self,
user_id: Union[int, str],
chat_id: Optional[Union[str, int]] = None,
user_id: int,
chat_id: Optional[int] = None,
message_id: Optional[int] = None,
inline_message_id: Optional[str] = None,
*,
@ -1781,7 +1781,7 @@ class ExtBot(Bot, Generic[RLARGS]):
async def get_user_profile_photos(
self,
user_id: Union[str, int],
user_id: int,
offset: Optional[int] = None,
limit: Optional[int] = None,
*,
@ -2032,7 +2032,7 @@ class ExtBot(Bot, Generic[RLARGS]):
async def promote_chat_member(
self,
chat_id: Union[str, int],
user_id: Union[str, int],
user_id: int,
can_change_info: Optional[bool] = None,
can_post_messages: Optional[bool] = None,
can_edit_messages: Optional[bool] = None,
@ -2100,7 +2100,7 @@ class ExtBot(Bot, Generic[RLARGS]):
async def restrict_chat_member(
self,
chat_id: Union[str, int],
user_id: Union[str, int],
user_id: int,
permissions: ChatPermissions,
until_date: Optional[Union[int, datetime]] = None,
use_independent_chat_permissions: Optional[bool] = None,
@ -2397,11 +2397,11 @@ class ExtBot(Bot, Generic[RLARGS]):
async def send_game(
self,
chat_id: Union[int, str],
chat_id: int,
game_short_name: str,
disable_notification: DVInput[bool] = DEFAULT_NONE,
reply_to_message_id: Optional[int] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
allow_sending_without_reply: ODVInput[bool] = DEFAULT_NONE,
protect_content: ODVInput[bool] = DEFAULT_NONE,
message_thread_id: Optional[int] = None,
@ -2450,7 +2450,7 @@ class ExtBot(Bot, Generic[RLARGS]):
is_flexible: Optional[bool] = None,
disable_notification: DVInput[bool] = DEFAULT_NONE,
reply_to_message_id: Optional[int] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
provider_data: Optional[Union[str, object]] = None,
send_phone_number_to_provider: Optional[bool] = None,
send_email_to_provider: Optional[bool] = None,
@ -2958,7 +2958,7 @@ class ExtBot(Bot, Generic[RLARGS]):
async def set_chat_administrator_custom_title(
self,
chat_id: Union[int, str],
user_id: Union[int, str],
user_id: int,
custom_title: str,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -3115,9 +3115,9 @@ class ExtBot(Bot, Generic[RLARGS]):
async def set_game_score(
self,
user_id: Union[int, str],
user_id: int,
score: int,
chat_id: Optional[Union[str, int]] = None,
chat_id: Optional[int] = None,
message_id: Optional[int] = None,
inline_message_id: Optional[str] = None,
force: Optional[bool] = None,
@ -3193,8 +3193,8 @@ class ExtBot(Bot, Generic[RLARGS]):
async def set_passport_data_errors(
self,
user_id: Union[str, int],
errors: Sequence[PassportElementError],
user_id: int,
errors: Sequence["PassportElementError"],
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -3238,7 +3238,7 @@ class ExtBot(Bot, Generic[RLARGS]):
async def set_sticker_set_thumbnail(
self,
name: str,
user_id: Union[str, int],
user_id: int,
thumbnail: Optional[FileInput] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -3296,7 +3296,7 @@ class ExtBot(Bot, Generic[RLARGS]):
chat_id: Optional[Union[str, int]] = None,
message_id: Optional[int] = None,
inline_message_id: Optional[str] = None,
reply_markup: Optional[InlineKeyboardMarkup] = None,
reply_markup: Optional["InlineKeyboardMarkup"] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -3320,7 +3320,7 @@ class ExtBot(Bot, Generic[RLARGS]):
async def unban_chat_member(
self,
chat_id: Union[str, int],
user_id: Union[str, int],
user_id: int,
only_if_banned: Optional[bool] = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -3449,7 +3449,7 @@ class ExtBot(Bot, Generic[RLARGS]):
async def upload_sticker_file(
self,
user_id: Union[str, int],
user_id: int,
sticker: Optional[FileInput],
sticker_format: Optional[str],
*,

2
tests/README.rst
View file

@ -72,7 +72,7 @@ complete and correct. To run it, export an environment variable first:
$ export TEST_OFFICIAL=true
and then run ``pytest tests/test_official.py``.
and then run ``pytest tests/test_official.py``. Note: You need py 3.10+ to run this test.
We also have another marker, ``@pytest.mark.dev``, which you can add to tests that you want to run selectively.
Use as follows:

13
tests/_passport/test_passportelementerrorfiles.py
View file

@ -19,6 +19,7 @@
import pytest
from telegram import PassportElementErrorFiles, PassportElementErrorSelfie
from telegram.warnings import PTBDeprecationWarning
from tests.auxil.slots import mro_slots
@ -58,11 +59,11 @@ class TestPassportElementErrorFilesWithoutRequest(TestPassportElementErrorFilesB
assert isinstance(passport_element_error_files_dict, dict)
assert passport_element_error_files_dict["source"] == passport_element_error_files.source
assert passport_element_error_files_dict["type"] == passport_element_error_files.type
assert passport_element_error_files_dict["message"] == passport_element_error_files.message
assert (
passport_element_error_files_dict["file_hashes"]
== passport_element_error_files.file_hashes
)
assert passport_element_error_files_dict["message"] == passport_element_error_files.message
def test_equality(self):
a = PassportElementErrorFiles(self.type_, self.file_hashes, self.message)
@ -87,3 +88,13 @@ class TestPassportElementErrorFilesWithoutRequest(TestPassportElementErrorFilesB
assert a != f
assert hash(a) != hash(f)
def test_file_hashes_deprecated(self, passport_element_error_files, recwarn):
passport_element_error_files.file_hashes
assert len(recwarn) == 1
assert (
"The attribute `file_hashes` will return a tuple instead of a list in future major"
" versions." in str(recwarn[0].message)
)
assert recwarn[0].category is PTBDeprecationWarning
assert recwarn[0].filename == __file__

19
tests/_passport/test_passportelementerrortranslationfiles.py
View file

@ -19,6 +19,7 @@
import pytest
from telegram import PassportElementErrorSelfie, PassportElementErrorTranslationFiles
from telegram.warnings import PTBDeprecationWarning
from tests.auxil.slots import mro_slots
@ -68,14 +69,14 @@ class TestPassportElementErrorTranslationFilesWithoutRequest(
passport_element_error_translation_files_dict["type"]
== passport_element_error_translation_files.type
)
assert (
passport_element_error_translation_files_dict["file_hashes"]
== passport_element_error_translation_files.file_hashes
)
assert (
passport_element_error_translation_files_dict["message"]
== passport_element_error_translation_files.message
)
assert (
passport_element_error_translation_files_dict["file_hashes"]
== passport_element_error_translation_files.file_hashes
)
def test_equality(self):
a = PassportElementErrorTranslationFiles(self.type_, self.file_hashes, self.message)
@ -100,3 +101,13 @@ class TestPassportElementErrorTranslationFilesWithoutRequest(
assert a != f
assert hash(a) != hash(f)
def test_file_hashes_deprecated(self, passport_element_error_translation_files, recwarn):
passport_element_error_translation_files.file_hashes
assert len(recwarn) == 1
assert (
"The attribute `file_hashes` will return a tuple instead of a list in future major"
" versions." in str(recwarn[0].message)
)
assert recwarn[0].category is PTBDeprecationWarning
assert recwarn[0].filename == __file__

11
tests/_passport/test_passportfile.py
View file

@ -19,6 +19,7 @@
import pytest
from telegram import Bot, File, PassportElementError, PassportFile
from telegram.warnings import PTBDeprecationWarning
from tests.auxil.bot_method_checks import (
check_defaults_handling,
check_shortcut_call,
@ -88,6 +89,16 @@ class TestPassportFileWithoutRequest(TestPassportFileBase):
assert a != e
assert hash(a) != hash(e)
def test_file_date_deprecated(self, passport_file, recwarn):
passport_file.file_date
assert len(recwarn) == 1
assert (
"The attribute `file_date` will return a datetime instead of an integer in future"
" major versions." in str(recwarn[0].message)
)
assert recwarn[0].category is PTBDeprecationWarning
assert recwarn[0].filename == __file__
async def test_get_file_instance_method(self, monkeypatch, passport_file):
async def make_assertion(*_, **kwargs):
result = kwargs["file_id"] == passport_file.file_id

1
tests/auxil/envvars.py
View file

@ -29,3 +29,4 @@ def env_var_2_bool(env_var: object) -> bool:
GITHUB_ACTION = os.getenv("GITHUB_ACTION", "")
TEST_WITH_OPT_DEPS = env_var_2_bool(os.getenv("TEST_WITH_OPT_DEPS", "true"))
RUN_TEST_OFFICIAL = env_var_2_bool(os.getenv("TEST_OFFICIAL"))

12
tests/conftest.py
View file

@ -18,6 +18,7 @@
# along with this program. If not, see [http://www.gnu.org/licenses/].
import asyncio
import datetime
import logging
import sys
from typing import Dict, List
from uuid import uuid4
@ -40,7 +41,7 @@ from telegram.ext.filters import MessageFilter, UpdateFilter
from tests.auxil.build_messages import DATE
from tests.auxil.ci_bots import BOT_INFO_PROVIDER
from tests.auxil.constants import PRIVATE_KEY
from tests.auxil.envvars import TEST_WITH_OPT_DEPS
from tests.auxil.envvars import RUN_TEST_OFFICIAL, TEST_WITH_OPT_DEPS
from tests.auxil.files import data_file
from tests.auxil.networking import NonchalantHttpxRequest
from tests.auxil.pytest_classes import PytestApplication, PytestBot, make_bot
@ -50,6 +51,15 @@ if TEST_WITH_OPT_DEPS:
import pytz
# Don't collect `test_official.py` on Python 3.10- since it uses newer features like X | Y syntax.
# Docs: https://docs.pytest.org/en/7.1.x/example/pythoncollection.html#customizing-test-collection
collect_ignore = []
if sys.version_info < (3, 10):
if RUN_TEST_OFFICIAL:
logging.warning("Skipping test_official.py since it requires Python 3.10+")
collect_ignore.append("test_official.py")
# This is here instead of in setup.cfg due to https://github.com/pytest-dev/pytest/issues/8343
def pytest_runtestloop(session: pytest.Session):
session.add_marker(

283
tests/test_official.py
View file

@ -17,17 +17,20 @@
# 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 os
import re
from typing import Dict, List, Set
from datetime import datetime
from types import FunctionType
from typing import Any, Callable, ForwardRef, Sequence, get_args, get_origin
import httpx
import pytest
from bs4 import BeautifulSoup
from bs4 import BeautifulSoup, PageElement, Tag
import telegram
from telegram._utils.defaultvalue import DefaultValue
from tests.auxil.envvars import env_var_2_bool
from telegram._utils.types import DVInput, FileInput, ODVInput
from telegram.ext import Defaults
from tests.auxil.envvars import RUN_TEST_OFFICIAL
IGNORED_OBJECTS = ("ResponseParameters", "CallbackGame")
GLOBALLY_IGNORED_PARAMETERS = {
@ -61,8 +64,42 @@ PTB_EXTRA_PARAMS = {
"InputFile": {"attach", "filename", "obj"},
}
# Types for certain parameters accepted by PTB but not in the official API
ADDITIONAL_TYPES = {
"photo": ForwardRef("PhotoSize"),
"video": ForwardRef("Video"),
"video_note": ForwardRef("VideoNote"),
"audio": ForwardRef("Audio"),
"document": ForwardRef("Document"),
"animation": ForwardRef("Animation"),
"voice": ForwardRef("Voice"),
"sticker": ForwardRef("Sticker"),
}
def _get_params_base(object_name: str, search_dict: Dict[str, Set[str]]) -> Set[str]:
# Exceptions to the "Array of" types, where we accept more types than the official API
# key: parameter name, value: type which must be present in the annotation
ARRAY_OF_EXCEPTIONS = {
"results": "InlineQueryResult", # + Callable
"commands": "BotCommand", # + tuple[str, str]
"keyboard": "KeyboardButton", # + sequence[sequence[str]]
# TODO: Deprecated and will be corrected (and removed) in next major PTB version:
"file_hashes": "list[str]",
}
# Special cases for other parameters that accept more types than the official API, and are
# too complex to compare/predict with official API:
EXCEPTIONS = { # (param_name, is_class): reduced form of annotation
("correct_option_id", False): int, # actual: Literal
("file_id", False): str, # actual: Union[str, objs_with_file_id_attr]
("invite_link", False): str, # actual: Union[str, ChatInviteLink]
("provider_data", False): str, # actual: Union[str, obj]
("callback_data", True): str, # actual: Union[str, obj]
("media", True): str, # actual: Union[str, InputMedia*, FileInput]
("data", True): str, # actual: Union[IdDocumentData, PersonalDetails, ResidentialAddress]
}
def _get_params_base(object_name: str, search_dict: dict[str, set[Any]]) -> set[Any]:
"""Helper function for the *_params functions below.
Given an object name and a search dict, goes through the keys of the search dict and checks if
the object name matches any of the regexes (keys). The union of all the sets (values) of the
@ -79,7 +116,7 @@ def _get_params_base(object_name: str, search_dict: Dict[str, Set[str]]) -> Set[
return out
def ptb_extra_params(object_name) -> Set[str]:
def ptb_extra_params(object_name: str) -> set[str]:
return _get_params_base(object_name, PTB_EXTRA_PARAMS)
@ -96,7 +133,7 @@ PTB_IGNORED_PARAMS = {
}
def ptb_ignored_params(object_name) -> Set[str]:
def ptb_ignored_params(object_name: str) -> set[str]:
return _get_params_base(object_name, PTB_IGNORED_PARAMS)
@ -111,22 +148,22 @@ IGNORED_PARAM_REQUIREMENTS = {
}
def ignored_param_requirements(object_name) -> Set[str]:
def ignored_param_requirements(object_name: str) -> set[str]:
return _get_params_base(object_name, IGNORED_PARAM_REQUIREMENTS)
# Arguments that are optional arguments for now for backwards compatibility
BACKWARDS_COMPAT_KWARGS = {}
BACKWARDS_COMPAT_KWARGS: dict[str, set[str]] = {}
def backwards_compat_kwargs(object_name: str) -> Set[str]:
def backwards_compat_kwargs(object_name: str) -> set[str]:
return _get_params_base(object_name, BACKWARDS_COMPAT_KWARGS)
IGNORED_PARAM_REQUIREMENTS.update(BACKWARDS_COMPAT_KWARGS)
def find_next_sibling_until(tag, name, until):
def find_next_sibling_until(tag: Tag, name: str, until: Tag) -> PageElement | None:
for sibling in tag.next_siblings:
if sibling is until:
return None
@ -135,7 +172,7 @@ def find_next_sibling_until(tag, name, until):
return None
def parse_table(h4) -> List[List[str]]:
def parse_table(h4: Tag) -> list[list[str]]:
"""Parses the Telegram doc table and has an output of a 2D list."""
table = find_next_sibling_until(h4, "table", h4.find_next_sibling("h4"))
if not table:
@ -143,9 +180,12 @@ def parse_table(h4) -> List[List[str]]:
return [[td.text for td in tr.find_all("td")] for tr in table.find_all("tr")[1:]]
def check_method(h4):
def check_method(h4: Tag) -> None:
name = h4.text # name of the method in telegram's docs.
method = getattr(telegram.Bot, name) # Retrieve our lib method
method: FunctionType | None = getattr(telegram.Bot, name, None) # Retrieve our lib method
if not method:
raise AssertionError(f"Method {name} not found in telegram.Bot")
table = parse_table(h4)
# Check arguments based on source
@ -159,7 +199,16 @@ def check_method(h4):
if param is None:
raise AssertionError(f"Parameter {tg_parameter[0]} not found in {method.__name__}")
# TODO: Check type via docstring
# Check if type annotation is present and correct
if param.annotation is inspect.Parameter.empty:
raise AssertionError(
f"Param {param.name!r} of {method.__name__!r} should have a type annotation"
)
if not check_param_type(param, tg_parameter, method):
raise AssertionError(
f"Param {param.name!r} of {method.__name__!r} should be {tg_parameter[1]}"
)
# Now check if the parameter is required or not
if not check_required_param(tg_parameter, param, method.__name__):
raise AssertionError(
@ -195,7 +244,7 @@ def check_method(h4):
)
def check_object(h4):
def check_object(h4: Tag) -> None:
name = h4.text
obj = getattr(telegram, name)
table = parse_table(h4)
@ -217,7 +266,15 @@ def check_object(h4):
param = sig.parameters.get(field)
if param is None:
raise AssertionError(f"Attribute {field} not found in {obj.__name__}")
# TODO: Check type via docstring
# Check if type annotation is present and correct
if param.annotation is inspect.Parameter.empty:
raise AssertionError(
f"Param {param.name!r} of {obj.__name__!r} should have a type annotation"
)
if not check_param_type(param, tg_parameter, obj):
raise AssertionError(
f"Param {param.name!r} of {obj.__name__!r} should be {tg_parameter[1]}"
)
if not check_required_param(tg_parameter, param, obj.__name__):
raise AssertionError(f"{obj.__name__!r} parameter {param.name!r} requirement mismatch")
@ -244,7 +301,7 @@ def is_parameter_required_by_tg(field: str) -> bool:
def check_required_param(
param_desc: List[str], param: inspect.Parameter, method_or_obj_name: str
param_desc: list[str], param: inspect.Parameter, method_or_obj_name: str
) -> bool:
"""Checks if the method/class parameter is a required/optional param as per Telegram docs.
@ -264,11 +321,187 @@ def check_defaults_type(ptb_param: inspect.Parameter) -> bool:
return DefaultValue.get_value(ptb_param.default) is None
to_run = env_var_2_bool(os.getenv("TEST_OFFICIAL"))
argvalues = []
names = []
def check_param_type(
ptb_param: inspect.Parameter, tg_parameter: list[str], obj: FunctionType | type
) -> bool:
"""This function checks whether the type annotation of the parameter is the same as the one
specified in the official API. It also checks for some special cases where we accept more types
if to_run:
Args:
ptb_param (inspect.Parameter): The parameter object from our methods/classes
tg_parameter (list[str]): The table row corresponding to the parameter from official API.
obj (object): The object (method/class) that we are checking.
Returns:
:obj:`bool`: The boolean returned represents whether our parameter's type annotation is the
same as Telegram's or not.
"""
# In order to evaluate the type annotation, we need to first have a mapping of the types
# specified in the official API to our types. The keys are types in the column of official API.
TYPE_MAPPING: dict[str, set[Any]] = {
"Integer or String": {int | str},
"Integer": {int},
"String": {str},
r"Boolean|True": {bool},
r"Float(?: number)?": {float},
# Distinguishing 1D and 2D Sequences and finding the inner type is done later.
r"Array of (?:Array of )?[\w\,\s]*": {Sequence},
r"InputFile(?: or String)?": {FileInput},
}
tg_param_type: str = tg_parameter[1] # Type of parameter as specified in the docs
is_class = inspect.isclass(obj)
# Let's check for a match:
mapped: set[type] = _get_params_base(tg_param_type, TYPE_MAPPING)
# We should have a maximum of one match.
assert len(mapped) <= 1, f"More than one match found for {tg_param_type}"
if not mapped: # no match found, it's from telegram module
# it could be a list of objects, so let's check that:
objs = _extract_words(tg_param_type)
# We want to store both string version of class and the class obj itself. e.g. "InputMedia"
# and InputMedia because some annotations might be ForwardRefs.
if len(objs) >= 2: # We have to unionize the objects
mapped_type: tuple[Any, ...] = (_unionizer(objs, False), _unionizer(objs, True))
else:
mapped_type = (
getattr(telegram, tg_param_type), # This will fail if it's not from telegram mod
ForwardRef(tg_param_type),
tg_param_type, # for some reason, some annotations are just a string.
)
elif len(mapped) == 1:
mapped_type = mapped.pop()
# Resolve nested annotations to get inner types.
if (ptb_annotation := list(get_args(ptb_param.annotation))) == []:
ptb_annotation = ptb_param.annotation # if it's not nested, just use the annotation
if isinstance(ptb_annotation, list):
# Some cleaning:
# Remove 'Optional[...]' from the annotation if it's present. We do it this way since: 1)
# we already check if argument should be optional or not + type checkers will complain.
# 2) we want to check if our `obj` is same as API's `obj`, and since python evaluates
# `Optional[obj] != obj` we have to remove the Optional, so that we can compare the two.
if type(None) in ptb_annotation:
ptb_annotation.remove(type(None))
# Cleaning done... now let's put it back together.
# Join all the annotations back (i.e. Union)
ptb_annotation = _unionizer(ptb_annotation, False)
# Last step, we need to use get_origin to get the original type, since using get_args
# above will strip that out.
wrapped = get_origin(ptb_param.annotation)
if wrapped is not None:
# collections.abc.Sequence -> typing.Sequence
if "collections.abc.Sequence" in str(wrapped):
wrapped = Sequence
ptb_annotation = wrapped[ptb_annotation]
# We have put back our annotation together after removing the NoneType!
# Now let's do the checking, starting with "Array of ..." types.
if "Array of " in tg_param_type:
assert mapped_type is Sequence
# For exceptions just check if they contain the annotation
if ptb_param.name in ARRAY_OF_EXCEPTIONS:
return ARRAY_OF_EXCEPTIONS[ptb_param.name] in str(ptb_annotation)
pattern = r"Array of(?: Array of)? ([\w\,\s]*)"
obj_match: re.Match | None = re.search(pattern, tg_param_type) # extract obj from string
if obj_match is None:
raise AssertionError(f"Array of {tg_param_type} not found in {ptb_param.name}")
obj_str: str = obj_match.group(1)
# is obj a regular type like str?
array_of_mapped: set[type] = _get_params_base(obj_str, TYPE_MAPPING)
if len(array_of_mapped) == 0: # no match found, it's from telegram module
# it could be a list of objects, so let's check that:
objs = _extract_words(obj_str)
# let's unionize all the objects, with and without ForwardRefs.
unionized_objs: list[type] = [_unionizer(objs, True), _unionizer(objs, False)]
else:
unionized_objs = [array_of_mapped.pop()]
# This means it is Array of Array of [obj]
if "Array of Array of" in tg_param_type:
return any(Sequence[Sequence[o]] == ptb_annotation for o in unionized_objs)
# This means it is Array of [obj]
return any(mapped_type[o] == ptb_annotation for o in unionized_objs)
# Special case for when the parameter is a default value parameter
for name, _ in inspect.getmembers(Defaults, lambda x: isinstance(x, property)):
if name in ptb_param.name: # no strict == since we have a param: `explanation_parse_mode`
# Check if it's DVInput or ODVInput
for param_type in [DVInput, ODVInput]:
parsed = param_type[mapped_type]
if ptb_annotation == parsed:
return True
return False
# Special case for send_* methods where we accept more types than the official API:
if (
ptb_param.name in ADDITIONAL_TYPES
and not isinstance(mapped_type, tuple)
and obj.__name__.startswith("send")
):
mapped_type = mapped_type | ADDITIONAL_TYPES[ptb_param.name]
for (param_name, expected_class), exception_type in EXCEPTIONS.items():
if ptb_param.name == param_name and is_class is expected_class:
ptb_annotation = exception_type
# Special case for datetimes
if (
re.search(
r"""([_]+|\b) # check for word boundary or underscore
date # check for "date"
[^\w]*\b # optionally check for a word after 'date'
""",
ptb_param.name,
re.VERBOSE,
)
or "Unix time" in tg_parameter[-1]
):
# TODO: Remove this in v22 when it becomes a datetime
datetime_exceptions = {
"file_date",
}
if ptb_param.name in datetime_exceptions:
return True
# If it's a class, we only accept datetime as the parameter
mapped_type = datetime if is_class else mapped_type | datetime
# Final check for the basic types
if isinstance(mapped_type, tuple) and any(ptb_annotation == t for t in mapped_type):
return True
return mapped_type == ptb_annotation
def _extract_words(text: str) -> set[str]:
"""Extracts all words from a string, removing all punctuation and words like 'and' & 'or'."""
return set(re.sub(r"[^\w\s]", "", text).split()) - {"and", "or"}
def _unionizer(annotation: Sequence[Any] | set[Any], forward_ref: bool) -> Any:
"""Returns a union of all the types in the annotation. If forward_ref is True, it wraps the
annotation in a ForwardRef and then unionizes."""
union = None
for t in annotation:
if forward_ref:
t = ForwardRef(t) # noqa: PLW2901
elif not forward_ref and isinstance(t, str): # we have to import objects from lib
t = getattr(telegram, t) # noqa: PLW2901
union = t if union is None else union | t
return union
argvalues: list[tuple[Callable[[Tag], None], Tag]] = []
names: list[str] = []
if RUN_TEST_OFFICIAL:
argvalues = []
names = []
request = httpx.get("https://core.telegram.org/bots/api")
@ -278,8 +511,10 @@ if to_run:
# Methods and types don't have spaces in them, luckily all other sections of the docs do
# TODO: don't depend on that
if "-" not in thing["name"]:
h4 = thing.parent
h4: Tag | None = thing.parent
if h4 is None:
raise AssertionError("h4 is None")
# Is it a method
if h4.text[0].lower() == h4.text[0]:
argvalues.append((check_method, h4))
@ -289,7 +524,7 @@ if to_run:
names.append(h4.text)
@pytest.mark.skipif(not to_run, reason="test_official is not enabled")
@pytest.mark.skipif(not RUN_TEST_OFFICIAL, reason="test_official is not enabled")
@pytest.mark.parametrize(("method", "data"), argvalues=argvalues, ids=names)
def test_official(method, data):
method(data)