python/aiogram
1
0
Fork 0
mirror of https://github.com/aiogram/aiogram.git synced 2026-04-08 16:37:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge remote-tracking branch 'origin/dev-3.x' into dev-3.x-proxy

Browse source 
# Conflicts:
#	aiogram/api/client/session/aiohttp.py
#	aiogram/api/client/session/base.py
#	poetry.lock
This commit is contained in:
mpa 2020-04-20 19:17:00 +04:00
commit 921f82b873
209 changed files with 2977 additions and 758 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/tests.yml vendored
View file

@ -38,7 +38,7 @@ jobs:
- name: Lint code
run: |
poetry run flake8 aiogram test
poetry run mypy aiogram tests
poetry run mypy aiogram
- name: Run tests
run: |

4
Makefile
View file

@ -86,11 +86,11 @@ flake8-report:
.PHONY: mypy
mypy:
$(py) mypy aiogram tests
$(py) mypy aiogram
.PHONY: mypy-report
mypy-report:
$(py) mypy aiogram tests --html-report $(reports_dir)/typechecking
$(py) mypy aiogram --html-report $(reports_dir)/typechecking
.PHONY: lint
lint: isort black flake8 mypy

6
aiogram/__init__.py
View file

@ -3,6 +3,7 @@ from .api.client import session
from .api.client.bot import Bot
from .dispatcher import filters, handler
from .dispatcher.dispatcher import Dispatcher
from .dispatcher.middlewares.base import BaseMiddleware
from .dispatcher.router import Router
try:
@ -22,9 +23,10 @@ __all__ = (
"session",
"Dispatcher",
"Router",
"BaseMiddleware",
"filters",
"handler",
)
__version__ = "3.0.0a2"
__api_version__ = "4.6"
__version__ = "3.0.0a3"
__api_version__ = "4.7"

89
aiogram/api/client/base.py
View file

@ -1,89 +0,0 @@
from __future__ import annotations
from contextlib import asynccontextmanager
from typing import Any, Optional, TypeVar
from ...utils.mixins import ContextInstanceMixin, DataMixin
from ...utils.token import extract_bot_id, validate_token
from ..methods import TelegramMethod
from .session.aiohttp import AiohttpSession
from .session.base import BaseSession
T = TypeVar("T")
class BaseBot(ContextInstanceMixin, DataMixin):
"""
Base class for bots
"""
def __init__(
self, token: str, session: BaseSession = None, parse_mode: Optional[str] = None
) -> None:
validate_token(token)
if session is None:
session = AiohttpSession()
self.session = session
self.parse_mode = parse_mode
self.__token = token
@property
def id(self) -> int:
"""
Get bot ID from token
:return:
"""
return extract_bot_id(self.__token)
async def __call__(self, method: TelegramMethod[T]) -> T:
"""
Call API method
:param method:
:return:
"""
return await self.session.make_request(self.__token, method)
async def close(self) -> None:
"""
Close bot session
"""
await self.session.close()
@asynccontextmanager
async def context(self, auto_close: bool = True):
"""
Generate bot context
:param auto_close:
:return:
"""
token = self.set_current(self)
try:
yield self
finally:
if auto_close:
await self.close()
self.reset_current(token)
def __hash__(self) -> int:
"""
Get hash for the token
:return:
"""
return hash(self.__token)
def __eq__(self, other: Any) -> bool:
"""
Compare current bot with another bot instance
:param other:
:return:
"""
if not isinstance(other, BaseBot):
return False
return hash(self) == hash(other)

237
aiogram/api/client/bot.py
View file

@ -1,8 +1,13 @@
from __future__ import annotations
import datetime
from typing import List, Optional, Union
from contextlib import asynccontextmanager
from typing import Any, AsyncIterator, List, Optional, TypeVar, Union
from async_lru import alru_cache
from ...utils.mixins import ContextInstanceMixin
from ...utils.token import extract_bot_id, validate_token
from ..methods import (
AddStickerToSet,
AnswerCallbackQuery,
@ -29,6 +34,7 @@ from ..methods import (
GetFile,
GetGameHighScores,
GetMe,
GetMyCommands,
GetStickerSet,
GetUpdates,
GetUserProfilePhotos,
@ -42,6 +48,7 @@ from ..methods import (
SendAudio,
SendChatAction,
SendContact,
SendDice,
SendDocument,
SendGame,
SendInvoice,
@ -62,16 +69,20 @@ from ..methods import (
SetChatStickerSet,
SetChatTitle,
SetGameScore,
SetMyCommands,
SetPassportDataErrors,
SetStickerPositionInSet,
SetStickerSetThumb,
SetWebhook,
StopMessageLiveLocation,
StopPoll,
TelegramMethod,
UnbanChatMember,
UnpinChatMessage,
UploadStickerFile,
)
from ..types import (
BotCommand,
Chat,
ChatMember,
ChatPermissions,
@ -98,18 +109,92 @@ from ..types import (
UserProfilePhotos,
WebhookInfo,
)
from .base import BaseBot
from .session.aiohttp import AiohttpSession
from .session.base import BaseSession
T = TypeVar("T")
class Bot(BaseBot):
class Bot(ContextInstanceMixin["Bot"]):
"""
Class where located all API methods
Main bot class
"""
@alru_cache()
def __init__(
self, token: str, session: Optional[BaseSession] = None, parse_mode: Optional[str] = None
) -> None:
validate_token(token)
if session is None:
session = AiohttpSession()
self.session = session
self.parse_mode = parse_mode
self.__token = token
@property
def id(self) -> int:
"""
Get bot ID from token
:return:
"""
return extract_bot_id(self.__token)
@asynccontextmanager
async def context(self, auto_close: bool = True) -> AsyncIterator[Bot]:
"""
Generate bot context
:param auto_close:
:return:
"""
token = self.set_current(self)
try:
yield self
finally:
if auto_close:
await self.close()
self.reset_current(token)
@alru_cache() # type: ignore
async def me(self) -> User:
return await self.get_me()
async def close(self) -> None:
"""
Close bot session
"""
await self.session.close()
async def __call__(self, method: TelegramMethod[T]) -> T:
"""
Call API method
:param method:
:return:
"""
return await self.session.make_request(self.__token, method)
def __hash__(self) -> int:
"""
Get hash for the token
:return:
"""
return hash(self.__token)
def __eq__(self, other: Any) -> bool:
"""
Compare current bot with another bot instance
:param other:
:return:
"""
if not isinstance(other, Bot):
return False
return hash(self) == hash(other)
# =============================================================================================
# Group: Getting updates
# Source: https://core.telegram.org/bots/api#getting-updates
@ -269,7 +354,7 @@ class Bot(BaseBot):
:param chat_id: Unique identifier for the target chat or username of the target channel
(in the format @channelusername)
:param text: Text of the message to be sent
:param text: Text of the message to be sent, 1-4096 characters after entities parsing
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in your bot's message.
:param disable_web_page_preview: Disables link previews for links in this message
@ -345,7 +430,7 @@ class Bot(BaseBot):
get a photo from the Internet, or upload a new photo using
multipart/form-data.
:param caption: Photo caption (may also be used when resending photos by file_id), 0-1024
characters
characters after entities parsing
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in the media caption.
:param disable_notification: Sends the message silently. Users will receive a notification
@ -398,7 +483,7 @@ class Bot(BaseBot):
exists on the Telegram servers (recommended), pass an HTTP URL as a String
for Telegram to get an audio file from the Internet, or upload a new one
using multipart/form-data.
:param caption: Audio caption, 0-1024 characters
:param caption: Audio caption, 0-1024 characters after entities parsing
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in the media caption.
:param duration: Duration of the audio in seconds
@ -468,7 +553,7 @@ class Bot(BaseBot):
can pass 'attach://<file_attach_name>' if the thumbnail was uploaded using
multipart/form-data under <file_attach_name>.
:param caption: Document caption (may also be used when resending documents by file_id),
0-1024 characters
0-1024 characters after entities parsing
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in the media caption.
:param disable_notification: Sends the message silently. Users will receive a notification
@ -532,7 +617,7 @@ class Bot(BaseBot):
can pass 'attach://<file_attach_name>' if the thumbnail was uploaded using
multipart/form-data under <file_attach_name>.
:param caption: Video caption (may also be used when resending videos by file_id), 0-1024
characters
characters after entities parsing
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in the media caption.
:param supports_streaming: Pass True, if the uploaded video is suitable for streaming
@ -600,7 +685,7 @@ class Bot(BaseBot):
can pass 'attach://<file_attach_name>' if the thumbnail was uploaded using
multipart/form-data under <file_attach_name>.
:param caption: Animation caption (may also be used when resending animation by file_id),
0-1024 characters
0-1024 characters after entities parsing
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in the media caption.
:param disable_notification: Sends the message silently. Users will receive a notification
@ -641,7 +726,7 @@ class Bot(BaseBot):
) -> Message:
"""
Use this method to send audio files, if you want Telegram clients to display the file as a
playable voice message. For this to work, your audio must be in an .ogg file encoded with
playable voice message. For this to work, your audio must be in an .OGG file encoded with
OPUS (other formats may be sent as Audio or Document). On success, the sent Message is
returned. Bots can currently send voice messages of up to 50 MB in size, this limit may be
changed in the future.
@ -654,7 +739,7 @@ class Bot(BaseBot):
the Telegram servers (recommended), pass an HTTP URL as a String for
Telegram to get a file from the Internet, or upload a new one using
multipart/form-data.
:param caption: Voice message caption, 0-1024 characters
:param caption: Voice message caption, 0-1024 characters after entities parsing
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in the media caption.
:param duration: Duration of the voice message in seconds
@ -1031,6 +1116,40 @@ class Bot(BaseBot):
)
return await self(call)
async def send_dice(
self,
chat_id: Union[int, str],
disable_notification: Optional[bool] = None,
reply_to_message_id: Optional[int] = None,
reply_markup: Optional[
Union[InlineKeyboardMarkup, ReplyKeyboardMarkup, ReplyKeyboardRemove, ForceReply]
] = None,
) -> Message:
"""
Use this method to send a dice, which will have a random value from 1 to 6. On success,
the sent Message is returned. (Yes, we're aware of the 'proper' singular of die. But it's
awkward, and we decided to help it change. One dice at a time!)
Source: https://core.telegram.org/bots/api#senddice
:param chat_id: Unique identifier for the target chat or username of the target channel
(in the format @channelusername)
:param disable_notification: Sends the message silently. Users will receive a notification
with no sound.
:param reply_to_message_id: If the message is a reply, ID of the original message
:param reply_markup: Additional interface options. A JSON-serialized object for an inline
keyboard, custom reply keyboard, instructions to remove reply
keyboard or to force a reply from the user.
:return: On success, the sent Message is returned.
"""
call = SendDice(
chat_id=chat_id,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
reply_markup=reply_markup,
)
return await self(call)
async def send_chat_action(self, chat_id: Union[int, str], action: str,) -> bool:
"""
Use this method when you need to tell the user that something is happening on the bot's
@ -1541,6 +1660,31 @@ class Bot(BaseBot):
)
return await self(call)
async def set_my_commands(self, commands: List[BotCommand],) -> bool:
"""
Use this method to change the list of the bot's commands. Returns True on success.
Source: https://core.telegram.org/bots/api#setmycommands
:param commands: A JSON-serialized list of bot commands to be set as the list of the bot's
commands. At most 100 commands can be specified.
:return: Returns True on success.
"""
call = SetMyCommands(commands=commands,)
return await self(call)
async def get_my_commands(self,) -> List[BotCommand]:
"""
Use this method to get the current list of the bot's commands. Requires no parameters.
Returns Array of BotCommand on success.
Source: https://core.telegram.org/bots/api#getmycommands
:return: Returns Array of BotCommand on success.
"""
call = GetMyCommands()
return await self(call)
# =============================================================================================
# Group: Updating messages
# Source: https://core.telegram.org/bots/api#updating-messages
@ -1562,7 +1706,7 @@ class Bot(BaseBot):
Source: https://core.telegram.org/bots/api#editmessagetext
:param text: New text of the message
:param text: New text of the message, 1-4096 characters after entities parsing
:param chat_id: Required if inline_message_id is not specified. Unique identifier for the
target chat or username of the target channel (in the format
@channelusername)
@ -1610,7 +1754,7 @@ class Bot(BaseBot):
message to edit
:param inline_message_id: Required if chat_id and message_id are not specified. Identifier
of the inline message
:param caption: New caption of the message
:param caption: New caption of the message, 0-1024 characters after entities parsing
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in the media caption.
:param reply_markup: A JSON-serialized object for an inline keyboard.
@ -1724,6 +1868,8 @@ class Bot(BaseBot):
Use this method to delete a message, including service messages, with the following
limitations:
- A message can only be deleted if it was sent less than 48 hours ago.
- A dice message in a private chat can only be deleted if it was sent more than 24 hours
ago.
- Bots can delete outgoing messages in private chats, groups, and supergroups.
- Bots can delete incoming messages in private chats.
- Bots granted can_post_messages permissions can delete outgoing messages in channels.
@ -1767,7 +1913,7 @@ class Bot(BaseBot):
(in the format @channelusername)
:param sticker: Sticker to send. Pass a file_id as String to send a file that exists on
the Telegram servers (recommended), pass an HTTP URL as a String for
Telegram to get a .webp file from the Internet, or upload a new one using
Telegram to get a .WEBP file from the Internet, or upload a new one using
multipart/form-data.
:param disable_notification: Sends the message silently. Users will receive a notification
with no sound.
@ -1800,7 +1946,7 @@ class Bot(BaseBot):
async def upload_sticker_file(self, user_id: int, png_sticker: InputFile,) -> File:
"""
Use this method to upload a .png file with a sticker for later use in createNewStickerSet
Use this method to upload a .PNG file with a sticker for later use in createNewStickerSet
and addStickerToSet methods (can be used multiple times). Returns the uploaded File on
success.
@ -1820,14 +1966,16 @@ class Bot(BaseBot):
user_id: int,
name: str,
title: str,
png_sticker: Union[InputFile, str],
emojis: str,
png_sticker: Optional[Union[InputFile, str]] = None,
tgs_sticker: Optional[InputFile] = None,
contains_masks: Optional[bool] = None,
mask_position: Optional[MaskPosition] = None,
) -> bool:
"""
Use this method to create new sticker set owned by a user. The bot will be able to edit
the created sticker set. Returns True on success.
Use this method to create a new sticker set owned by a user. The bot will be able to edit
the sticker set thus created. You must use exactly one of the fields png_sticker or
tgs_sticker. Returns True on success.
Source: https://core.telegram.org/bots/api#createnewstickerset
@ -1837,13 +1985,16 @@ class Bot(BaseBot):
begin with a letter, can't contain consecutive underscores and must end in
'_by_<bot username>'. <bot_username> is case insensitive. 1-64 characters.
:param title: Sticker set title, 1-64 characters
:param png_sticker: Png image with the sticker, must be up to 512 kilobytes in size,
:param emojis: One or more emoji corresponding to the sticker
:param png_sticker: PNG image with the sticker, must be up to 512 kilobytes in size,
dimensions must not exceed 512px, and either width or height must be
exactly 512px. Pass a file_id as a String to send a file that already
exists on the Telegram servers, pass an HTTP URL as a String for
Telegram to get a file from the Internet, or upload a new one using
multipart/form-data.
:param emojis: One or more emoji corresponding to the sticker
:param tgs_sticker: TGS animation with the sticker, uploaded using multipart/form-data.
See https://core.telegram.org/animated_stickers#technical-requirements
for technical requirements
:param contains_masks: Pass True, if a set of mask stickers should be created
:param mask_position: A JSON-serialized object for position where the mask should be
placed on faces
@ -1853,8 +2004,9 @@ class Bot(BaseBot):
user_id=user_id,
name=name,
title=title,
png_sticker=png_sticker,
emojis=emojis,
png_sticker=png_sticker,
tgs_sticker=tgs_sticker,
contains_masks=contains_masks,
mask_position=mask_position,
)
@ -1866,22 +2018,29 @@ class Bot(BaseBot):
name: str,
png_sticker: Union[InputFile, str],
emojis: str,
tgs_sticker: Optional[InputFile] = None,
mask_position: Optional[MaskPosition] = None,
) -> bool:
"""
Use this method to add a new sticker to a set created by the bot. Returns True on success.
Use this method to add a new sticker to a set created by the bot. You must use exactly one
of the fields png_sticker or tgs_sticker. Animated stickers can be added to animated
sticker sets and only to them. Animated sticker sets can have up to 50 stickers. Static
sticker sets can have up to 120 stickers. Returns True on success.
Source: https://core.telegram.org/bots/api#addstickertoset
:param user_id: User identifier of sticker set owner
:param name: Sticker set name
:param png_sticker: Png image with the sticker, must be up to 512 kilobytes in size,
:param png_sticker: PNG image with the sticker, must be up to 512 kilobytes in size,
dimensions must not exceed 512px, and either width or height must be
exactly 512px. Pass a file_id as a String to send a file that already
exists on the Telegram servers, pass an HTTP URL as a String for
Telegram to get a file from the Internet, or upload a new one using
multipart/form-data.
:param emojis: One or more emoji corresponding to the sticker
:param tgs_sticker: TGS animation with the sticker, uploaded using multipart/form-data.
See https://core.telegram.org/animated_stickers#technical-requirements
for technical requirements
:param mask_position: A JSON-serialized object for position where the mask should be
placed on faces
:return: Returns True on success.
@ -1891,13 +2050,14 @@ class Bot(BaseBot):
name=name,
png_sticker=png_sticker,
emojis=emojis,
tgs_sticker=tgs_sticker,
mask_position=mask_position,
)
return await self(call)
async def set_sticker_position_in_set(self, sticker: str, position: int,) -> bool:
"""
Use this method to move a sticker in a set created by the bot to a specific position .
Use this method to move a sticker in a set created by the bot to a specific position.
Returns True on success.
Source: https://core.telegram.org/bots/api#setstickerpositioninset
@ -1922,6 +2082,31 @@ class Bot(BaseBot):
call = DeleteStickerFromSet(sticker=sticker,)
return await self(call)
async def set_sticker_set_thumb(
self, name: str, user_id: int, thumb: Optional[Union[InputFile, str]] = None,
) -> bool:
"""
Use this method to set the thumbnail of a sticker set. Animated thumbnails can be set for
animated sticker sets only. Returns True on success.
Source: https://core.telegram.org/bots/api#setstickersetthumb
:param name: Sticker set name
:param user_id: User identifier of the sticker set owner
:param thumb: A PNG image with the thumbnail, must be up to 128 kilobytes in size and have
width and height exactly 100px, or a TGS animation with the thumbnail up to
32 kilobytes in size; see
https://core.telegram.org/animated_stickers#technical-requirements for
animated sticker technical requirements. Pass a file_id as a String to send
a file that already exists on the Telegram servers, pass an HTTP URL as a
String for Telegram to get a file from the Internet, or upload a new one
using multipart/form-data.. Animated sticker set thumbnail can't be uploaded
via HTTP URL.
:return: Returns True on success.
"""
call = SetStickerSetThumb(name=name, user_id=user_id, thumb=thumb,)
return await self(call)
# =============================================================================================
# Group: Inline mode
# Source: https://core.telegram.org/bots/api#inline-mode

10
aiogram/api/client/session/aiohttp.py
View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import AsyncGenerator, Callable, Optional, TypeVar, Type, Tuple, Dict, Any, Union, cast
from typing import AsyncGenerator, Callable, Optional, TypeVar, Type, Tuple, Dict, Union, cast
from aiohttp import ClientSession, ClientTimeout, FormData, BasicAuth, TCPConnector
@ -16,8 +16,8 @@ class AiohttpSession(BaseSession[_ProxyType]):
def __init__(
self,
api: TelegramAPIServer = PRODUCTION,
json_loads: Optional[Callable] = None,
json_dumps: Optional[Callable] = None,
json_loads: Optional[Callable[..., str]] = None,
json_dumps: Optional[Callable[..., str]] = None,
proxy: Optional[_ProxyType] = None,
):
super(AiohttpSession, self).__init__(
@ -64,9 +64,7 @@ class AiohttpSession(BaseSession[_ProxyType]):
async def create_session(self) -> ClientSession:
if self._session is None or self._session.closed:
self._session = ClientSession(
connector=self._connector_type(**self._connector_init)
)
self._session = ClientSession()
return self._session

16
aiogram/api/client/session/base.py
View file

@ -3,7 +3,8 @@ from __future__ import annotations
import abc
import datetime
import json
from typing import Any, AsyncGenerator, Callable, Optional, TypeVar, Union, Generic
from types import TracebackType
from typing import Any, AsyncGenerator, Callable, Optional, Type, TypeVar, Union, Generic
from aiogram.utils.exceptions import TelegramAPIError
@ -18,8 +19,8 @@ class BaseSession(abc.ABC, Generic[_ProxyType]):
def __init__(
self,
api: Optional[TelegramAPIServer] = None,
json_loads: Optional[Callable[[Any], Any]] = None,
json_dumps: Optional[Callable[[Any], Any]] = None,
json_loads: Optional[Callable[..., str]] = None,
json_dumps: Optional[Callable[..., str]] = None,
proxy: Optional[_ProxyType] = None,
) -> None:
if api is None:
@ -40,7 +41,7 @@ class BaseSession(abc.ABC, Generic[_ProxyType]):
raise TelegramAPIError(response.description)
@abc.abstractmethod
async def close(self): # pragma: no cover
async def close(self) -> None: # pragma: no cover
pass
@abc.abstractmethod
@ -76,5 +77,10 @@ class BaseSession(abc.ABC, Generic[_ProxyType]):
async def __aenter__(self) -> BaseSession:
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
async def __aexit__(
self,
exc_type: Optional[Type[BaseException]],
exc_value: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
await self.close()

8
aiogram/api/methods/__init__.py
View file

@ -24,6 +24,7 @@ from .get_chat_members_count import GetChatMembersCount
from .get_file import GetFile
from .get_game_high_scores import GetGameHighScores
from .get_me import GetMe
from .get_my_commands import GetMyCommands
from .get_sticker_set import GetStickerSet
from .get_updates import GetUpdates
from .get_user_profile_photos import GetUserProfilePhotos
@ -37,6 +38,7 @@ from .send_animation import SendAnimation
from .send_audio import SendAudio
from .send_chat_action import SendChatAction
from .send_contact import SendContact
from .send_dice import SendDice
from .send_document import SendDocument
from .send_game import SendGame
from .send_invoice import SendInvoice
@ -57,8 +59,10 @@ from .set_chat_photo import SetChatPhoto
from .set_chat_sticker_set import SetChatStickerSet
from .set_chat_title import SetChatTitle
from .set_game_score import SetGameScore
from .set_my_commands import SetMyCommands
from .set_passport_data_errors import SetPassportDataErrors
from .set_sticker_position_in_set import SetStickerPositionInSet
from .set_sticker_set_thumb import SetStickerSetThumb
from .set_webhook import SetWebhook
from .stop_message_live_location import StopMessageLiveLocation
from .stop_poll import StopPoll
@ -91,6 +95,7 @@ __all__ = (
"SendVenue",
"SendContact",
"SendPoll",
"SendDice",
"SendChatAction",
"GetUserProfilePhotos",
"GetFile",
@ -115,6 +120,8 @@ __all__ = (
"SetChatStickerSet",
"DeleteChatStickerSet",
"AnswerCallbackQuery",
"SetMyCommands",
"GetMyCommands",
"EditMessageText",
"EditMessageCaption",
"EditMessageMedia",
@ -128,6 +135,7 @@ __all__ = (
"AddStickerToSet",
"SetStickerPositionInSet",
"DeleteStickerFromSet",
"SetStickerSetThumb",
"AnswerInlineQuery",
"SendInvoice",
"AnswerShippingQuery",

14
aiogram/api/methods/add_sticker_to_set.py
View file

@ -6,7 +6,10 @@ from .base import Request, TelegramMethod, prepare_file
class AddStickerToSet(TelegramMethod[bool]):
"""
Use this method to add a new sticker to a set created by the bot. Returns True on success.
Use this method to add a new sticker to a set created by the bot. You must use exactly one of
the fields png_sticker or tgs_sticker. Animated stickers can be added to animated sticker sets
and only to them. Animated sticker sets can have up to 50 stickers. Static sticker sets can
have up to 120 stickers. Returns True on success.
Source: https://core.telegram.org/bots/api#addstickertoset
"""
@ -18,19 +21,24 @@ class AddStickerToSet(TelegramMethod[bool]):
name: str
"""Sticker set name"""
png_sticker: Union[InputFile, str]
"""Png image with the sticker, must be up to 512 kilobytes in size, dimensions must not exceed
"""PNG image with the sticker, must be up to 512 kilobytes in size, dimensions must not exceed
512px, and either width or height must be exactly 512px. Pass a file_id as a String to send
a file that already exists on the Telegram servers, pass an HTTP URL as a String for
Telegram to get a file from the Internet, or upload a new one using multipart/form-data."""
emojis: str
"""One or more emoji corresponding to the sticker"""
tgs_sticker: Optional[InputFile] = None
"""TGS animation with the sticker, uploaded using multipart/form-data. See
https://core.telegram.org/animated_stickers#technical-requirements for technical
requirements"""
mask_position: Optional[MaskPosition] = None
"""A JSON-serialized object for position where the mask should be placed on faces"""
def build_request(self) -> Request:
data: Dict[str, Any] = self.dict(exclude={"png_sticker"})
data: Dict[str, Any] = self.dict(exclude={"png_sticker", "tgs_sticker"})
files: Dict[str, InputFile] = {}
prepare_file(data=data, files=files, name="png_sticker", value=self.png_sticker)
prepare_file(data=data, files=files, name="tgs_sticker", value=self.tgs_sticker)
return Request(method="addStickerToSet", data=data, files=files)

12
aiogram/api/methods/base.py
View file

@ -2,7 +2,7 @@ from __future__ import annotations
import abc
import secrets
from typing import TYPE_CHECKING, Any, Dict, Generic, Optional, Type, TypeVar, Union
from typing import TYPE_CHECKING, Any, Dict, Generator, Generic, Optional, TypeVar, Union
from pydantic import BaseConfig, BaseModel, Extra
from pydantic.generics import GenericModel
@ -24,7 +24,7 @@ class Request(BaseModel):
class Config(BaseConfig):
arbitrary_types_allowed = True
def render_webhook_request(self):
def render_webhook_request(self) -> Dict[str, Any]:
return {
"method": self.method,
**{key: value for key, value in self.data.items() if value is not None},
@ -48,7 +48,7 @@ class TelegramMethod(abc.ABC, BaseModel, Generic[T]):
@property
@abc.abstractmethod
def __returning__(self) -> Type: # pragma: no cover
def __returning__(self) -> type: # pragma: no cover
pass
@abc.abstractmethod
@ -62,14 +62,14 @@ class TelegramMethod(abc.ABC, BaseModel, Generic[T]):
async def emit(self, bot: Bot) -> T:
return await bot(self)
def __await__(self):
def __await__(self) -> Generator[Any, None, T]:
from aiogram.api.client.bot import Bot
bot = Bot.get_current(no_error=False)
return self.emit(bot).__await__()
def prepare_file(name: str, value: Any, data: Dict[str, Any], files: Dict[str, Any]):
def prepare_file(name: str, value: Any, data: Dict[str, Any], files: Dict[str, Any]) -> None:
if not value:
return
if name == "thumb":
@ -101,7 +101,7 @@ def prepare_media_file(data: Dict[str, Any], files: Dict[str, InputFile]) -> Non
and isinstance(data["media"]["media"], InputFile)
):
tag = secrets.token_urlsafe(10)
files[tag] = data["media"].pop("media") # type: ignore
files[tag] = data["media"].pop("media")
data["media"]["media"] = f"attach://{tag}"

20
aiogram/api/methods/create_new_sticker_set.py
View file

@ -6,8 +6,9 @@ from .base import Request, TelegramMethod, prepare_file
class CreateNewStickerSet(TelegramMethod[bool]):
"""
Use this method to create new sticker set owned by a user. The bot will be able to edit the
created sticker set. Returns True on success.
Use this method to create a new sticker set owned by a user. The bot will be able to edit the
sticker set thus created. You must use exactly one of the fields png_sticker or tgs_sticker.
Returns True on success.
Source: https://core.telegram.org/bots/api#createnewstickerset
"""
@ -23,22 +24,27 @@ class CreateNewStickerSet(TelegramMethod[bool]):
case insensitive. 1-64 characters."""
title: str
"""Sticker set title, 1-64 characters"""
png_sticker: Union[InputFile, str]
"""Png image with the sticker, must be up to 512 kilobytes in size, dimensions must not exceed
emojis: str
"""One or more emoji corresponding to the sticker"""
png_sticker: Optional[Union[InputFile, str]] = None
"""PNG image with the sticker, must be up to 512 kilobytes in size, dimensions must not exceed
512px, and either width or height must be exactly 512px. Pass a file_id as a String to send
a file that already exists on the Telegram servers, pass an HTTP URL as a String for
Telegram to get a file from the Internet, or upload a new one using multipart/form-data."""
emojis: str
"""One or more emoji corresponding to the sticker"""
tgs_sticker: Optional[InputFile] = None
"""TGS animation with the sticker, uploaded using multipart/form-data. See
https://core.telegram.org/animated_stickers#technical-requirements for technical
requirements"""
contains_masks: Optional[bool] = None
"""Pass True, if a set of mask stickers should be created"""
mask_position: Optional[MaskPosition] = None
"""A JSON-serialized object for position where the mask should be placed on faces"""
def build_request(self) -> Request:
data: Dict[str, Any] = self.dict(exclude={"png_sticker"})
data: Dict[str, Any] = self.dict(exclude={"png_sticker", "tgs_sticker"})
files: Dict[str, InputFile] = {}
prepare_file(data=data, files=files, name="png_sticker", value=self.png_sticker)
prepare_file(data=data, files=files, name="tgs_sticker", value=self.tgs_sticker)
return Request(method="createNewStickerSet", data=data, files=files)

1
aiogram/api/methods/delete_message.py
View file

@ -8,6 +8,7 @@ class DeleteMessage(TelegramMethod[bool]):
Use this method to delete a message, including service messages, with the following
limitations:
- A message can only be deleted if it was sent less than 48 hours ago.
- A dice message in a private chat can only be deleted if it was sent more than 24 hours ago.
- Bots can delete outgoing messages in private chats, groups, and supergroups.
- Bots can delete incoming messages in private chats.
- Bots granted can_post_messages permissions can delete outgoing messages in channels.

2
aiogram/api/methods/edit_message_caption.py
View file

@ -22,7 +22,7 @@ class EditMessageCaption(TelegramMethod[Union[Message, bool]]):
inline_message_id: Optional[str] = None
"""Required if chat_id and message_id are not specified. Identifier of the inline message"""
caption: Optional[str] = None
"""New caption of the message"""
"""New caption of the message, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/methods/edit_message_text.py
View file

@ -15,7 +15,7 @@ class EditMessageText(TelegramMethod[Union[Message, bool]]):
__returning__ = Union[Message, bool]
text: str
"""New text of the message"""
"""New text of the message, 1-4096 characters after entities parsing"""
chat_id: Optional[Union[int, str]] = None
"""Required if inline_message_id is not specified. Unique identifier for the target chat or
username of the target channel (in the format @channelusername)"""

20
aiogram/api/methods/get_my_commands.py Normal file
View file

@ -0,0 +1,20 @@
from typing import Any, Dict, List
from ..types import BotCommand
from .base import Request, TelegramMethod
class GetMyCommands(TelegramMethod[List[BotCommand]]):
"""
Use this method to get the current list of the bot's commands. Requires no parameters. Returns
Array of BotCommand on success.
Source: https://core.telegram.org/bots/api#getmycommands
"""
__returning__ = List[BotCommand]
def build_request(self) -> Request:
data: Dict[str, Any] = self.dict()
return Request(method="getMyCommands", data=data)

3
aiogram/api/methods/send_animation.py
View file

@ -43,7 +43,8 @@ class SendAnimation(TelegramMethod[Message]):
file, so you can pass 'attach://<file_attach_name>' if the thumbnail was uploaded using
multipart/form-data under <file_attach_name>."""
caption: Optional[str] = None
"""Animation caption (may also be used when resending animation by file_id), 0-1024 characters"""
"""Animation caption (may also be used when resending animation by file_id), 0-1024 characters
after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/methods/send_audio.py
View file

@ -32,7 +32,7 @@ class SendAudio(TelegramMethod[Message]):
Telegram servers (recommended), pass an HTTP URL as a String for Telegram to get an audio
file from the Internet, or upload a new one using multipart/form-data."""
caption: Optional[str] = None
"""Audio caption, 0-1024 characters"""
"""Audio caption, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

40
aiogram/api/methods/send_dice.py Normal file
View file

@ -0,0 +1,40 @@
from typing import Any, Dict, Optional, Union
from ..types import (
ForceReply,
InlineKeyboardMarkup,
Message,
ReplyKeyboardMarkup,
ReplyKeyboardRemove,
)
from .base import Request, TelegramMethod
class SendDice(TelegramMethod[Message]):
"""
Use this method to send a dice, which will have a random value from 1 to 6. On success, the
sent Message is returned. (Yes, we're aware of the 'proper' singular of die. But it's awkward,
and we decided to help it change. One dice at a time!)
Source: https://core.telegram.org/bots/api#senddice
"""
__returning__ = Message
chat_id: Union[int, str]
"""Unique identifier for the target chat or username of the target channel (in the format
@channelusername)"""
disable_notification: Optional[bool] = None
"""Sends the message silently. Users will receive a notification with no sound."""
reply_to_message_id: Optional[int] = None
"""If the message is a reply, ID of the original message"""
reply_markup: Optional[
Union[InlineKeyboardMarkup, ReplyKeyboardMarkup, ReplyKeyboardRemove, ForceReply]
] = None
"""Additional interface options. A JSON-serialized object for an inline keyboard, custom reply
keyboard, instructions to remove reply keyboard or to force a reply from the user."""
def build_request(self) -> Request:
data: Dict[str, Any] = self.dict()
return Request(method="sendDice", data=data)

3
aiogram/api/methods/send_document.py
View file

@ -37,7 +37,8 @@ class SendDocument(TelegramMethod[Message]):
file, so you can pass 'attach://<file_attach_name>' if the thumbnail was uploaded using
multipart/form-data under <file_attach_name>."""
caption: Optional[str] = None
"""Document caption (may also be used when resending documents by file_id), 0-1024 characters"""
"""Document caption (may also be used when resending documents by file_id), 0-1024 characters
after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/methods/send_message.py
View file

@ -23,7 +23,7 @@ class SendMessage(TelegramMethod[Message]):
"""Unique identifier for the target chat or username of the target channel (in the format
@channelusername)"""
text: str
"""Text of the message to be sent"""
"""Text of the message to be sent, 1-4096 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in your bot's message."""

3
aiogram/api/methods/send_photo.py
View file

@ -28,7 +28,8 @@ class SendPhoto(TelegramMethod[Message]):
(recommended), pass an HTTP URL as a String for Telegram to get a photo from the Internet,
or upload a new photo using multipart/form-data."""
caption: Optional[str] = None
"""Photo caption (may also be used when resending photos by file_id), 0-1024 characters"""
"""Photo caption (may also be used when resending photos by file_id), 0-1024 characters after
entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/methods/send_poll.py
View file

@ -25,7 +25,7 @@ class SendPoll(TelegramMethod[Message]):
question: str
"""Poll question, 1-255 characters"""
options: List[str]
"""List of answer options, 2-10 strings 1-100 characters each"""
"""A JSON-serialized list of answer options, 2-10 strings 1-100 characters each"""
is_anonymous: Optional[bool] = None
"""True, if the poll needs to be anonymous, defaults to True"""
type: Optional[str] = None

2
aiogram/api/methods/send_sticker.py
View file

@ -26,7 +26,7 @@ class SendSticker(TelegramMethod[Message]):
@channelusername)"""
sticker: Union[InputFile, str]
"""Sticker to send. Pass a file_id as String to send a file that exists on the Telegram
servers (recommended), pass an HTTP URL as a String for Telegram to get a .webp file from
servers (recommended), pass an HTTP URL as a String for Telegram to get a .WEBP file from
the Internet, or upload a new one using multipart/form-data."""
disable_notification: Optional[bool] = None
"""Sends the message silently. Users will receive a notification with no sound."""

3
aiogram/api/methods/send_video.py
View file

@ -43,7 +43,8 @@ class SendVideo(TelegramMethod[Message]):
file, so you can pass 'attach://<file_attach_name>' if the thumbnail was uploaded using
multipart/form-data under <file_attach_name>."""
caption: Optional[str] = None
"""Video caption (may also be used when resending videos by file_id), 0-1024 characters"""
"""Video caption (may also be used when resending videos by file_id), 0-1024 characters after
entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

4
aiogram/api/methods/send_voice.py
View file

@ -14,7 +14,7 @@ from .base import Request, TelegramMethod, prepare_file
class SendVoice(TelegramMethod[Message]):
"""
Use this method to send audio files, if you want Telegram clients to display the file as a
playable voice message. For this to work, your audio must be in an .ogg file encoded with OPUS
playable voice message. For this to work, your audio must be in an .OGG file encoded with OPUS
(other formats may be sent as Audio or Document). On success, the sent Message is returned.
Bots can currently send voice messages of up to 50 MB in size, this limit may be changed in
the future.
@ -32,7 +32,7 @@ class SendVoice(TelegramMethod[Message]):
servers (recommended), pass an HTTP URL as a String for Telegram to get a file from the
Internet, or upload a new one using multipart/form-data."""
caption: Optional[str] = None
"""Voice message caption, 0-1024 characters"""
"""Voice message caption, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

23
aiogram/api/methods/set_my_commands.py Normal file
View file

@ -0,0 +1,23 @@
from typing import Any, Dict, List
from ..types import BotCommand
from .base import Request, TelegramMethod
class SetMyCommands(TelegramMethod[bool]):
"""
Use this method to change the list of the bot's commands. Returns True on success.
Source: https://core.telegram.org/bots/api#setmycommands
"""
__returning__ = bool
commands: List[BotCommand]
"""A JSON-serialized list of bot commands to be set as the list of the bot's commands. At most
100 commands can be specified."""
def build_request(self) -> Request:
data: Dict[str, Any] = self.dict()
return Request(method="setMyCommands", data=data)

2
aiogram/api/methods/set_sticker_position_in_set.py
View file

@ -5,7 +5,7 @@ from .base import Request, TelegramMethod
class SetStickerPositionInSet(TelegramMethod[bool]):
"""
Use this method to move a sticker in a set created by the bot to a specific position . Returns
Use this method to move a sticker in a set created by the bot to a specific position. Returns
True on success.
Source: https://core.telegram.org/bots/api#setstickerpositioninset

36
aiogram/api/methods/set_sticker_set_thumb.py Normal file
View file

@ -0,0 +1,36 @@
from typing import Any, Dict, Optional, Union
from ..types import InputFile
from .base import Request, TelegramMethod, prepare_file
class SetStickerSetThumb(TelegramMethod[bool]):
"""
Use this method to set the thumbnail of a sticker set. Animated thumbnails can be set for
animated sticker sets only. Returns True on success.
Source: https://core.telegram.org/bots/api#setstickersetthumb
"""
__returning__ = bool
name: str
"""Sticker set name"""
user_id: int
"""User identifier of the sticker set owner"""
thumb: Optional[Union[InputFile, str]] = None
"""A PNG image with the thumbnail, must be up to 128 kilobytes in size and have width and
height exactly 100px, or a TGS animation with the thumbnail up to 32 kilobytes in size; see
https://core.telegram.org/animated_stickers#technical-requirements for animated sticker
technical requirements. Pass a file_id as a String to send a file that already exists on
the Telegram servers, pass an HTTP URL as a String for Telegram to get a file from the
Internet, or upload a new one using multipart/form-data.. Animated sticker set thumbnail
can't be uploaded via HTTP URL."""
def build_request(self) -> Request:
data: Dict[str, Any] = self.dict(exclude={"thumb"})
files: Dict[str, InputFile] = {}
prepare_file(data=data, files=files, name="thumb", value=self.thumb)
return Request(method="setStickerSetThumb", data=data, files=files)

2
aiogram/api/methods/upload_sticker_file.py
View file

@ -6,7 +6,7 @@ from .base import Request, TelegramMethod, prepare_file
class UploadStickerFile(TelegramMethod[File]):
"""
Use this method to upload a .png file with a sticker for later use in createNewStickerSet and
Use this method to upload a .PNG file with a sticker for later use in createNewStickerSet and
addStickerToSet methods (can be used multiple times). Returns the uploaded File on success.
Source: https://core.telegram.org/bots/api#uploadstickerfile

4
aiogram/api/types/__init__.py
View file

@ -1,6 +1,7 @@
from .animation import Animation
from .audio import Audio
from .base import TelegramObject
from .bot_command import BotCommand
from .callback_game import CallbackGame
from .callback_query import CallbackQuery
from .chat import Chat
@ -9,6 +10,7 @@ from .chat_permissions import ChatPermissions
from .chat_photo import ChatPhoto
from .chosen_inline_result import ChosenInlineResult
from .contact import Contact
from .dice import Dice
from .document import Document
from .encrypted_credentials import EncryptedCredentials
from .encrypted_passport_element import EncryptedPassportElement
@ -121,6 +123,7 @@ __all__ = (
"PollOption",
"PollAnswer",
"Poll",
"Dice",
"UserProfilePhotos",
"File",
"ReplyKeyboardMarkup",
@ -135,6 +138,7 @@ __all__ = (
"ChatPhoto",
"ChatMember",
"ChatPermissions",
"BotCommand",
"ResponseParameters",
"InputMedia",
"InputMediaPhoto",

2
aiogram/api/types/base.py
View file

@ -5,7 +5,7 @@ from pydantic import BaseModel, Extra
from aiogram.utils.mixins import ContextInstanceMixin
class TelegramObject(ContextInstanceMixin, BaseModel):
class TelegramObject(ContextInstanceMixin["TelegramObject"], BaseModel):
class Config:
use_enum_values = True
orm_mode = True

17
aiogram/api/types/bot_command.py Normal file
View file

@ -0,0 +1,17 @@
from __future__ import annotations
from .base import MutableTelegramObject
class BotCommand(MutableTelegramObject):
"""
This object represents a bot command.
Source: https://core.telegram.org/bots/api#botcommand
"""
command: str
"""Text of the command, 1-32 characters. Can contain only lowercase English letters, digits
and underscores."""
description: str
"""Description of the command, 3-256 characters."""

15
aiogram/api/types/dice.py Normal file
View file

@ -0,0 +1,15 @@
from __future__ import annotations
from .base import TelegramObject
class Dice(TelegramObject):
"""
This object represents a dice with random value from 1 to 6. (Yes, we're aware of the 'proper'
singular of die. But it's awkward, and we decided to help it change. One dice at a time!)
Source: https://core.telegram.org/bots/api#dice
"""
value: int
"""Value of the dice, 1-6"""

2
aiogram/api/types/inline_query_result_audio.py
View file

@ -31,7 +31,7 @@ class InlineQueryResultAudio(InlineQueryResult):
title: str
"""Title"""
caption: Optional[str] = None
"""Caption, 0-1024 characters"""
"""Caption, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/inline_query_result_cached_audio.py
View file

@ -29,7 +29,7 @@ class InlineQueryResultCachedAudio(InlineQueryResult):
audio_file_id: str
"""A valid file identifier for the audio file"""
caption: Optional[str] = None
"""Caption, 0-1024 characters"""
"""Caption, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/inline_query_result_cached_document.py
View file

@ -33,7 +33,7 @@ class InlineQueryResultCachedDocument(InlineQueryResult):
description: Optional[str] = None
"""Short description of the result"""
caption: Optional[str] = None
"""Caption of the document to be sent, 0-1024 characters"""
"""Caption of the document to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/inline_query_result_cached_gif.py
View file

@ -29,7 +29,7 @@ class InlineQueryResultCachedGif(InlineQueryResult):
title: Optional[str] = None
"""Title for the result"""
caption: Optional[str] = None
"""Caption of the GIF file to be sent, 0-1024 characters"""
"""Caption of the GIF file to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/inline_query_result_cached_mpeg4_gif.py
View file

@ -30,7 +30,7 @@ class InlineQueryResultCachedMpeg4Gif(InlineQueryResult):
title: Optional[str] = None
"""Title for the result"""
caption: Optional[str] = None
"""Caption of the MPEG-4 file to be sent, 0-1024 characters"""
"""Caption of the MPEG-4 file to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/inline_query_result_cached_photo.py
View file

@ -31,7 +31,7 @@ class InlineQueryResultCachedPhoto(InlineQueryResult):
description: Optional[str] = None
"""Short description of the result"""
caption: Optional[str] = None
"""Caption of the photo to be sent, 0-1024 characters"""
"""Caption of the photo to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/inline_query_result_cached_video.py
View file

@ -31,7 +31,7 @@ class InlineQueryResultCachedVideo(InlineQueryResult):
description: Optional[str] = None
"""Short description of the result"""
caption: Optional[str] = None
"""Caption of the video to be sent, 0-1024 characters"""
"""Caption of the video to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/inline_query_result_cached_voice.py
View file

@ -31,7 +31,7 @@ class InlineQueryResultCachedVoice(InlineQueryResult):
title: str
"""Voice message title"""
caption: Optional[str] = None
"""Caption, 0-1024 characters"""
"""Caption, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/inline_query_result_document.py
View file

@ -34,7 +34,7 @@ class InlineQueryResultDocument(InlineQueryResult):
mime_type: str
"""Mime type of the content of the file, either 'application/pdf' or 'application/zip'"""
caption: Optional[str] = None
"""Caption of the document to be sent, 0-1024 characters"""
"""Caption of the document to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/inline_query_result_gif.py
View file

@ -37,7 +37,7 @@ class InlineQueryResultGif(InlineQueryResult):
title: Optional[str] = None
"""Title for the result"""
caption: Optional[str] = None
"""Caption of the GIF file to be sent, 0-1024 characters"""
"""Caption of the GIF file to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/inline_query_result_mpeg4_gif.py
View file

@ -38,7 +38,7 @@ class InlineQueryResultMpeg4Gif(InlineQueryResult):
title: Optional[str] = None
"""Title for the result"""
caption: Optional[str] = None
"""Caption of the MPEG-4 file to be sent, 0-1024 characters"""
"""Caption of the MPEG-4 file to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/inline_query_result_photo.py
View file

@ -37,7 +37,7 @@ class InlineQueryResultPhoto(InlineQueryResult):
description: Optional[str] = None
"""Short description of the result"""
caption: Optional[str] = None
"""Caption of the photo to be sent, 0-1024 characters"""
"""Caption of the photo to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/inline_query_result_video.py
View file

@ -35,7 +35,7 @@ class InlineQueryResultVideo(InlineQueryResult):
title: str
"""Title for the result"""
caption: Optional[str] = None
"""Caption of the video to be sent, 0-1024 characters"""
"""Caption of the video to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

4
aiogram/api/types/inline_query_result_voice.py
View file

@ -13,7 +13,7 @@ if TYPE_CHECKING: # pragma: no cover
class InlineQueryResultVoice(InlineQueryResult):
"""
Represents a link to a voice recording in an .ogg container encoded with OPUS. By default,
Represents a link to a voice recording in an .OGG container encoded with OPUS. By default,
this voice recording will be sent by the user. Alternatively, you can use
input_message_content to send a message with the specified content instead of the the voice
message.
@ -32,7 +32,7 @@ class InlineQueryResultVoice(InlineQueryResult):
title: str
"""Recording title"""
caption: Optional[str] = None
"""Caption, 0-1024 characters"""
"""Caption, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

8
aiogram/api/types/input_file.py
View file

@ -4,7 +4,7 @@ import io
import os
from abc import ABC, abstractmethod
from pathlib import Path
from typing import AsyncGenerator, Optional, Union
from typing import AsyncGenerator, AsyncIterator, Iterator, Optional, Union
import aiofiles as aiofiles
@ -24,14 +24,14 @@ class InputFile(ABC):
self.chunk_size = chunk_size
@classmethod
def __get_validators__(cls):
yield
def __get_validators__(cls) -> Iterator[None]:
yield None
@abstractmethod
async def read(self, chunk_size: int) -> AsyncGenerator[bytes, None]: # pragma: no cover
yield b""
async def __aiter__(self):
async def __aiter__(self) -> AsyncIterator[bytes]:
async for chunk in self.read(self.chunk_size):
yield chunk

2
aiogram/api/types/input_media_animation.py
View file

@ -32,7 +32,7 @@ class InputMediaAnimation(InputMedia):
file, so you can pass 'attach://<file_attach_name>' if the thumbnail was uploaded using
multipart/form-data under <file_attach_name>."""
caption: Optional[str] = None
"""Caption of the animation to be sent, 0-1024 characters"""
"""Caption of the animation to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/input_media_audio.py
View file

@ -32,7 +32,7 @@ class InputMediaAudio(InputMedia):
file, so you can pass 'attach://<file_attach_name>' if the thumbnail was uploaded using
multipart/form-data under <file_attach_name>."""
caption: Optional[str] = None
"""Caption of the audio to be sent, 0-1024 characters"""
"""Caption of the audio to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/input_media_document.py
View file

@ -32,7 +32,7 @@ class InputMediaDocument(InputMedia):
file, so you can pass 'attach://<file_attach_name>' if the thumbnail was uploaded using
multipart/form-data under <file_attach_name>."""
caption: Optional[str] = None
"""Caption of the document to be sent, 0-1024 characters"""
"""Caption of the document to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/input_media_photo.py
View file

@ -25,7 +25,7 @@ class InputMediaPhoto(InputMedia):
'attach://<file_attach_name>' to upload a new one using multipart/form-data under
<file_attach_name> name."""
caption: Optional[str] = None
"""Caption of the photo to be sent, 0-1024 characters"""
"""Caption of the photo to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

2
aiogram/api/types/input_media_video.py
View file

@ -32,7 +32,7 @@ class InputMediaVideo(InputMedia):
file, so you can pass 'attach://<file_attach_name>' if the thumbnail was uploaded using
multipart/form-data under <file_attach_name>."""
caption: Optional[str] = None
"""Caption of the video to be sent, 0-1024 characters"""
"""Caption of the video to be sent, 0-1024 characters after entities parsing"""
parse_mode: Optional[str] = None
"""Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or
inline URLs in the media caption."""

59
aiogram/api/types/message.py
View file

@ -13,6 +13,7 @@ if TYPE_CHECKING: # pragma: no cover
from .audio import Audio
from .chat import Chat
from .contact import Contact
from .dice import Dice
from .document import Document
from .force_reply import ForceReply
from .game import Game
@ -49,6 +50,7 @@ if TYPE_CHECKING: # pragma: no cover
SendMessage,
SendPhoto,
SendPoll,
SendDice,
SendSticker,
SendVenue,
SendVideo,
@ -95,7 +97,7 @@ class Message(TelegramObject):
author_signature: Optional[str] = None
"""Signature of the post author for messages in channels"""
text: Optional[str] = None
"""For text messages, the actual UTF-8 text of the message, 0-4096 characters."""
"""For text messages, the actual UTF-8 text of the message, 0-4096 characters"""
entities: Optional[List[MessageEntity]] = None
"""For text messages, special entities like usernames, URLs, bot commands, etc. that appear in
the text"""
@ -131,6 +133,8 @@ class Message(TelegramObject):
"""Message is a venue, information about the venue"""
poll: Optional[Poll] = None
"""Message is a native poll, information about the poll"""
dice: Optional[Dice] = None
"""Message is a dice with random value from 1 to 6"""
new_chat_members: Optional[List[User]] = None
"""New members that were added to the group or supergroup and information about them (the bot
itself may be one of these members)"""
@ -181,7 +185,7 @@ class Message(TelegramObject):
buttons."""
@property
def content_type(self):
def content_type(self) -> str:
if self.text:
return ContentType.TEXT
if self.audio:
@ -236,6 +240,8 @@ class Message(TelegramObject):
return ContentType.PASSPORT_DATA
if self.poll:
return ContentType.POLL
if self.dice:
return ContentType.DICE
return ContentType.UNKNOWN
@ -1081,6 +1087,52 @@ class Message(TelegramObject):
reply_markup=reply_markup,
)
def reply_dice(
self,
disable_notification: Optional[bool] = None,
reply_markup: Optional[
Union[InlineKeyboardMarkup, ReplyKeyboardMarkup, ReplyKeyboardRemove, ForceReply]
] = None,
) -> SendDice:
"""
Reply with dice
:param disable_notification:
:param reply_markup:
:return:
"""
from ..methods import SendDice
return SendDice(
chat_id=self.chat.id,
disable_notification=disable_notification,
reply_to_message_id=self.message_id,
reply_markup=reply_markup,
)
def answer_dice(
self,
disable_notification: Optional[bool] = None,
reply_markup: Optional[
Union[InlineKeyboardMarkup, ReplyKeyboardMarkup, ReplyKeyboardRemove, ForceReply]
] = None,
) -> SendDice:
"""
Answer with dice
:param disable_notification:
:param reply_markup:
:return:
"""
from ..methods import SendDice
return SendDice(
chat_id=self.chat.id,
disable_notification=disable_notification,
reply_to_message_id=None,
reply_markup=reply_markup,
)
def reply_sticker(
self,
sticker: Union[InputFile, str],
@ -1479,7 +1531,8 @@ class ContentType(helper.Helper):
DELETE_CHAT_PHOTO = helper.Item() # delete_chat_photo
GROUP_CHAT_CREATED = helper.Item() # group_chat_created
PASSPORT_DATA = helper.Item() # passport_data
POLL = helper.Item()
POLL = helper.Item() # poll
DICE = helper.Item() # dice
UNKNOWN = helper.Item() # unknown
ANY = helper.Item() # any

2
aiogram/api/types/sticker.py
View file

@ -28,7 +28,7 @@ class Sticker(TelegramObject):
is_animated: bool
"""True, if the sticker is animated"""
thumb: Optional[PhotoSize] = None
"""Sticker thumbnail in the .webp or .jpg format"""
"""Sticker thumbnail in the .WEBP or .JPG format"""
emoji: Optional[str] = None
"""Emoji associated with the sticker"""
set_name: Optional[str] = None

5
aiogram/api/types/sticker_set.py
View file

@ -1,10 +1,11 @@
from __future__ import annotations
from typing import TYPE_CHECKING, List
from typing import TYPE_CHECKING, List, Optional
from .base import TelegramObject
if TYPE_CHECKING: # pragma: no cover
from .photo_size import PhotoSize
from .sticker import Sticker
@ -25,3 +26,5 @@ class StickerSet(TelegramObject):
"""True, if the sticker set contains masks"""
stickers: List[Sticker]
"""List of all set stickers"""
thumb: Optional[PhotoSize] = None
"""Sticker set thumbnail in the .WEBP or .TGS format"""

2
aiogram/api/types/user.py
View file

@ -32,7 +32,7 @@ class User(TelegramObject):
"""True, if the bot supports inline queries. Returned only in getMe."""
@property
def full_name(self):
def full_name(self) -> str:
if self.last_name:
return f"{self.first_name} {self.last_name}"
return self.first_name

12
aiogram/dispatcher/dispatcher.py
View file

@ -1,3 +1,5 @@
from __future__ import annotations
import asyncio
import contextvars
import warnings
@ -96,7 +98,7 @@ class Dispatcher(Router):
update_id = update.update_id + 1
@classmethod
async def _silent_call_request(cls, bot: Bot, result: TelegramMethod) -> None:
async def _silent_call_request(cls, bot: Bot, result: TelegramMethod[Any]) -> None:
"""
Simulate answer into WebHook
@ -172,7 +174,7 @@ class Dispatcher(Router):
raise
async def feed_webhook_update(
self, bot: Bot, update: Union[Update, Dict[str, Any]], _timeout: int = 55, **kwargs
self, bot: Bot, update: Union[Update, Dict[str, Any]], _timeout: int = 55, **kwargs: Any
) -> Optional[Dict[str, Any]]:
if not isinstance(update, Update): # Allow to use raw updates
update = Update(**update)
@ -181,18 +183,18 @@ class Dispatcher(Router):
loop = asyncio.get_running_loop()
waiter = loop.create_future()
def release_waiter(*args: Any):
def release_waiter(*args: Any) -> None:
if not waiter.done():
waiter.set_result(None)
timeout_handle = loop.call_later(_timeout, release_waiter)
process_updates: Future = asyncio.ensure_future(
process_updates: Future[Any] = asyncio.ensure_future(
self._feed_webhook_update(bot=bot, update=update, **kwargs)
)
process_updates.add_done_callback(release_waiter, context=ctx)
def process_response(task: Future):
def process_response(task: Future[Any]) -> None:
warnings.warn(
f"Detected slow response into webhook.\n"
f"Telegram is waiting for response only first 60 seconds and then re-send update.\n"

17
aiogram/dispatcher/event/handler.py
View file

@ -1,7 +1,7 @@
import inspect
from dataclasses import dataclass, field
from functools import partial
from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple, Union
from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple, Type, Union
from aiogram.dispatcher.filters.base import BaseFilter
from aiogram.dispatcher.handler.base import BaseHandler
@ -10,7 +10,7 @@ CallbackType = Callable[[Any], Awaitable[Any]]
SyncFilter = Callable[[Any], Any]
AsyncFilter = Callable[[Any], Awaitable[Any]]
FilterType = Union[SyncFilter, AsyncFilter, BaseFilter]
HandlerType = Union[FilterType, BaseHandler]
HandlerType = Union[FilterType, Type[BaseHandler]]
@dataclass
@ -19,20 +19,20 @@ class CallableMixin:
awaitable: bool = field(init=False)
spec: inspect.FullArgSpec = field(init=False)
def __post_init__(self):
def __post_init__(self) -> None:
callback = self.callback
self.awaitable = inspect.isawaitable(callback) or inspect.iscoroutinefunction(callback)
while hasattr(callback, "__wrapped__"): # Try to resolve decorated callbacks
callback = callback.__wrapped__
callback = callback.__wrapped__ # type: ignore
self.spec = inspect.getfullargspec(callback)
def _prepare_kwargs(self, kwargs):
def _prepare_kwargs(self, kwargs: Dict[str, Any]) -> Dict[str, Any]:
if self.spec.varkw:
return kwargs
return {k: v for k, v in kwargs.items() if k in self.spec.args}
async def call(self, *args, **kwargs):
async def call(self, *args: Any, **kwargs: Any) -> Any:
wrapped = partial(self.callback, *args, **self._prepare_kwargs(kwargs))
if self.awaitable:
return await wrapped()
@ -49,10 +49,9 @@ class HandlerObject(CallableMixin):
callback: HandlerType
filters: Optional[List[FilterObject]] = None
def __post_init__(self):
def __post_init__(self) -> None:
super(HandlerObject, self).__post_init__()
if inspect.isclass(self.callback) and issubclass(self.callback, BaseHandler):
if inspect.isclass(self.callback) and issubclass(self.callback, BaseHandler): # type: ignore
self.awaitable = True
async def check(self, *args: Any, **kwargs: Any) -> Tuple[bool, Dict[str, Any]]:

62
aiogram/dispatcher/event/observer.py
View file

@ -9,6 +9,7 @@ from typing import (
Dict,
Generator,
List,
NoReturn,
Optional,
Type,
)
@ -16,6 +17,7 @@ from typing import (
from pydantic import ValidationError
from ..filters.base import BaseFilter
from ..middlewares.types import MiddlewareStep, UpdateType
from .handler import CallbackType, FilterObject, FilterType, HandlerObject, HandlerType
if TYPE_CHECKING: # pragma: no cover
@ -26,6 +28,17 @@ class SkipHandler(Exception):
pass
class CancelHandler(Exception):
pass
def skip(message: Optional[str] = None) -> NoReturn:
"""
Raise an SkipHandler
"""
raise SkipHandler(message or "Event skipped")
class EventObserver:
"""
Base events observer
@ -95,10 +108,8 @@ class TelegramEventObserver(EventObserver):
"""
registry: List[Type[BaseFilter]] = []
router: Optional[Router] = self.router
while router:
for router in self.router.chain:
observer = router.observers[self.event_name]
router = router.parent_router
for filter_ in observer.filters:
if filter_ in registry:
@ -133,6 +144,37 @@ class TelegramEventObserver(EventObserver):
return filters
async def trigger_middleware(
self, step: MiddlewareStep, event: UpdateType, data: Dict[str, Any], result: Any = None,
) -> None:
"""
Trigger middlewares chain
:param step:
:param event:
:param data:
:param result:
:return:
"""
reverse = step == MiddlewareStep.POST_PROCESS
recursive = self.event_name == "update" or step == MiddlewareStep.PROCESS
if self.event_name == "update":
routers = self.router.chain
else:
routers = self.router.chain_head
for router in routers:
await router.middleware.trigger(
step=step,
event_name=self.event_name,
event=event,
data=data,
result=result,
reverse=reverse,
)
if not recursive:
break
def register(
self, callback: HandlerType, *filters: FilterType, **bound_filters: Any
) -> HandlerType:
@ -153,12 +195,24 @@ class TelegramEventObserver(EventObserver):
Propagate event to handlers and stops propagation on first match.
Handler will be called when all its filters is pass.
"""
event = args[0]
await self.trigger_middleware(step=MiddlewareStep.PRE_PROCESS, event=event, data=kwargs)
for handler in self.handlers:
result, data = await handler.check(*args, **kwargs)
if result:
kwargs.update(data)
await self.trigger_middleware(
step=MiddlewareStep.PROCESS, event=event, data=kwargs
)
try:
yield await handler.call(*args, **kwargs)
response = await handler.call(*args, **kwargs)
await self.trigger_middleware(
step=MiddlewareStep.POST_PROCESS,
event=event,
data=kwargs,
result=response,
)
yield response
except SkipHandler:
continue
break

8
aiogram/dispatcher/filters/__init__.py
View file

@ -1,8 +1,9 @@
from typing import Dict, Tuple, Union
from typing import Dict, Tuple, Type
from .base import BaseFilter
from .command import Command, CommandObject
from .content_types import ContentTypesFilter
from .exception import ExceptionMessageFilter, ExceptionTypeFilter
from .text import Text
__all__ = (
@ -12,9 +13,11 @@ __all__ = (
"Command",
"CommandObject",
"ContentTypesFilter",
"ExceptionMessageFilter",
"ExceptionTypeFilter",
)
BUILTIN_FILTERS: Dict[str, Union[Tuple[BaseFilter], Tuple]] = {
BUILTIN_FILTERS: Dict[str, Tuple[Type[BaseFilter], ...]] = {
"update": (),
"message": (Text, Command, ContentTypesFilter),
"edited_message": (Text, Command, ContentTypesFilter),
@ -27,4 +30,5 @@ BUILTIN_FILTERS: Dict[str, Union[Tuple[BaseFilter], Tuple]] = {
"pre_checkout_query": (),
"poll": (),
"poll_answer": (),
"error": (ExceptionMessageFilter, ExceptionTypeFilter),
}

7
aiogram/dispatcher/filters/base.py
View file

@ -1,5 +1,5 @@
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any, Dict, Union
from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Union
from pydantic import BaseModel
@ -9,14 +9,13 @@ class BaseFilter(ABC, BaseModel):
# This checking type-hint is needed because mypy checks validity of overrides and raises:
# error: Signature of "__call__" incompatible with supertype "BaseFilter" [override]
# https://mypy.readthedocs.io/en/latest/error_code_list.html#check-validity-of-overrides-override
pass
__call__: Callable[..., Awaitable[Union[bool, Dict[str, Any]]]]
else: # pragma: no cover
@abstractmethod
async def __call__(self, *args: Any, **kwargs: Any) -> Union[bool, Dict[str, Any]]:
pass
def __await__(self): # pragma: no cover
def __await__(self): # type: ignore # pragma: no cover
# Is needed only for inspection and this method is never be called
return self.__call__

2
aiogram/dispatcher/filters/command.py
View file

@ -10,7 +10,7 @@ from aiogram import Bot
from aiogram.api.types import Message
from aiogram.dispatcher.filters import BaseFilter
CommandPatterType = Union[str, re.Pattern] # type: ignore
CommandPatterType = Union[str, re.Pattern]
class Command(BaseFilter):

36
aiogram/dispatcher/filters/exception.py Normal file
View file

@ -0,0 +1,36 @@
import re
from typing import Any, Dict, Pattern, Tuple, Type, Union, cast
from pydantic import validator
from aiogram.dispatcher.filters import BaseFilter
class ExceptionTypeFilter(BaseFilter):
exception: Union[Type[Exception], Tuple[Type[Exception]]]
class Config:
arbitrary_types_allowed = True
async def __call__(self, exception: Exception) -> Union[bool, Dict[str, Any]]:
return isinstance(exception, self.exception)
class ExceptionMessageFilter(BaseFilter):
match: Union[str, Pattern[str]]
class Config:
arbitrary_types_allowed = True
@validator("match")
def _validate_match(cls, value: Union[str, Pattern[str]]) -> Union[str, Pattern[str]]:
if isinstance(value, str):
return re.compile(value)
return value
async def __call__(self, exception: Exception) -> Union[bool, Dict[str, Any]]:
pattern = cast(Pattern[str], self.match)
result = pattern.match(str(exception))
if not result:
return False
return {"match_exception": result}

2
aiogram/dispatcher/filters/text.py
View file

@ -80,7 +80,7 @@ class Text(BaseFilter):
# Impossible because the validator prevents this situation
return False # pragma: no cover
def prepare_text(self, text: str):
def prepare_text(self, text: str) -> str:
if self.text_ignore_case:
return str(text).lower()
else:

2
aiogram/dispatcher/handler/__init__.py
View file

@ -1,6 +1,7 @@
from .base import BaseHandler, BaseHandlerMixin
from .callback_query import CallbackQueryHandler
from .chosen_inline_result import ChosenInlineResultHandler
from .error import ErrorHandler
from .inline_query import InlineQueryHandler
from .message import MessageHandler, MessageHandlerCommandMixin
from .poll import PollHandler
@ -12,6 +13,7 @@ __all__ = (
"BaseHandlerMixin",
"CallbackQueryHandler",
"ChosenInlineResultHandler",
"ErrorHandler",
"InlineQueryHandler",
"MessageHandler",
"MessageHandlerCommandMixin",

10
aiogram/dispatcher/handler/base.py
View file

@ -1,5 +1,5 @@
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any, Dict, Generic, TypeVar
from typing import TYPE_CHECKING, Any, Dict, Generic, TypeVar, cast
from aiogram import Bot
from aiogram.api.types import Update
@ -25,16 +25,16 @@ class BaseHandler(BaseHandlerMixin[T], ABC):
@property
def bot(self) -> Bot:
if "bot" in self.data:
return self.data["bot"]
return Bot.get_current()
return cast(Bot, self.data["bot"])
return Bot.get_current(no_error=False)
@property
def update(self) -> Update:
return self.data["update"]
return cast(Update, self.data["update"])
@abstractmethod
async def handle(self) -> Any: # pragma: no cover
pass
def __await__(self):
def __await__(self) -> Any:
return self.handle().__await__()

17
aiogram/dispatcher/handler/error.py Normal file
View file

@ -0,0 +1,17 @@
from abc import ABC
from aiogram.dispatcher.handler.base import BaseHandler
class ErrorHandler(BaseHandler[Exception], ABC):
"""
Base class for errors handlers
"""
@property
def exception_name(self) -> str:
return self.event.__class__.__name__
@property
def exception_message(self) -> str:
return str(self.event)

4
aiogram/dispatcher/handler/message.py
View file

@ -1,5 +1,5 @@
from abc import ABC
from typing import Optional
from typing import Optional, cast
from aiogram.api.types import Chat, Message, User
from aiogram.dispatcher.filters import CommandObject
@ -24,5 +24,5 @@ class MessageHandlerCommandMixin(BaseHandlerMixin[Message]):
@property
def command(self) -> Optional[CommandObject]:
if "command" in self.data:
return self.data["command"]
return cast(CommandObject, self.data["command"])
return None

0
aiogram/dispatcher/middlewares/__init__.py Normal file
View file

61
aiogram/dispatcher/middlewares/abstract.py Normal file
View file

@ -0,0 +1,61 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any, Dict, Optional
from aiogram.dispatcher.middlewares.types import MiddlewareStep, UpdateType
if TYPE_CHECKING: # pragma: no cover
from aiogram.dispatcher.middlewares.manager import MiddlewareManager
class AbstractMiddleware(ABC):
"""
Abstract class for middleware.
"""
def __init__(self) -> None:
self._manager: Optional[MiddlewareManager] = None
@property
def manager(self) -> MiddlewareManager:
"""
Instance of MiddlewareManager
"""
if self._manager is None:
raise RuntimeError("Middleware is not configured!")
return self._manager
def setup(self, manager: MiddlewareManager, _stack_level: int = 1) -> AbstractMiddleware:
"""
Mark middleware as configured
:param manager:
:param _stack_level:
:return:
"""
if self.configured:
return manager.setup(self, _stack_level=_stack_level + 1)
self._manager = manager
return self
@property
def configured(self) -> bool:
"""
Check middleware is configured
:return:
"""
return bool(self._manager)
@abstractmethod
async def trigger(
self,
step: MiddlewareStep,
event_name: str,
event: UpdateType,
data: Dict[str, Any],
result: Any = None,
) -> Any: # pragma: no cover
pass

317
aiogram/dispatcher/middlewares/base.py Normal file
View file

@ -0,0 +1,317 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Dict
from aiogram.dispatcher.middlewares.abstract import AbstractMiddleware
from aiogram.dispatcher.middlewares.types import MiddlewareStep, UpdateType
if TYPE_CHECKING: # pragma: no cover
from aiogram.api.types import (
CallbackQuery,
ChosenInlineResult,
InlineQuery,
Message,
Poll,
PollAnswer,
PreCheckoutQuery,
ShippingQuery,
Update,
)
class BaseMiddleware(AbstractMiddleware):
"""
Base class for middleware.
All methods on the middle always must be coroutines and name starts with "on_" like "on_process_message".
"""
async def trigger(
self,
step: MiddlewareStep,
event_name: str,
event: UpdateType,
data: Dict[str, Any],
result: Any = None,
) -> Any:
"""
Trigger action.
:param step:
:param event_name:
:param event:
:param data:
:param result:
:return:
"""
handler_name = f"on_{step.value}_{event_name}"
handler = getattr(self, handler_name, None)
if not handler:
return None
args = (event, result, data) if step == MiddlewareStep.POST_PROCESS else (event, data)
return await handler(*args)
if TYPE_CHECKING: # pragma: no cover
# =============================================================================================
# Event that triggers before process <event>
# =============================================================================================
async def on_pre_process_update(self, update: Update, data: Dict[str, Any]) -> Any:
"""
Event that triggers before process update
"""
async def on_pre_process_message(self, message: Message, data: Dict[str, Any]) -> Any:
"""
Event that triggers before process message
"""
async def on_pre_process_edited_message(
self, edited_message: Message, data: Dict[str, Any]
) -> Any:
"""
Event that triggers before process edited_message
"""
async def on_pre_process_channel_post(
self, channel_post: Message, data: Dict[str, Any]
) -> Any:
"""
Event that triggers before process channel_post
"""
async def on_pre_process_edited_channel_post(
self, edited_channel_post: Message, data: Dict[str, Any]
) -> Any:
"""
Event that triggers before process edited_channel_post
"""
async def on_pre_process_inline_query(
self, inline_query: InlineQuery, data: Dict[str, Any]
) -> Any:
"""
Event that triggers before process inline_query
"""
async def on_pre_process_chosen_inline_result(
self, chosen_inline_result: ChosenInlineResult, data: Dict[str, Any]
) -> Any:
"""
Event that triggers before process chosen_inline_result
"""
async def on_pre_process_callback_query(
self, callback_query: CallbackQuery, data: Dict[str, Any]
) -> Any:
"""
Event that triggers before process callback_query
"""
async def on_pre_process_shipping_query(
self, shipping_query: ShippingQuery, data: Dict[str, Any]
) -> Any:
"""
Event that triggers before process shipping_query
"""
async def on_pre_process_pre_checkout_query(
self, pre_checkout_query: PreCheckoutQuery, data: Dict[str, Any]
) -> Any:
"""
Event that triggers before process pre_checkout_query
"""
async def on_pre_process_poll(self, poll: Poll, data: Dict[str, Any]) -> Any:
"""
Event that triggers before process poll
"""
async def on_pre_process_poll_answer(
self, poll_answer: PollAnswer, data: Dict[str, Any]
) -> Any:
"""
Event that triggers before process poll_answer
"""
async def on_pre_process_error(self, exception: Exception, data: Dict[str, Any]) -> Any:
"""
Event that triggers before process error
"""
# =============================================================================================
# Event that triggers on process <event> after filters.
# =============================================================================================
async def on_process_update(self, update: Update, data: Dict[str, Any]) -> Any:
"""
Event that triggers on process update
"""
async def on_process_message(self, message: Message, data: Dict[str, Any]) -> Any:
"""
Event that triggers on process message
"""
async def on_process_edited_message(
self, edited_message: Message, data: Dict[str, Any]
) -> Any:
"""
Event that triggers on process edited_message
"""
async def on_process_channel_post(
self, channel_post: Message, data: Dict[str, Any]
) -> Any:
"""
Event that triggers on process channel_post
"""
async def on_process_edited_channel_post(
self, edited_channel_post: Message, data: Dict[str, Any]
) -> Any:
"""
Event that triggers on process edited_channel_post
"""
async def on_process_inline_query(
self, inline_query: InlineQuery, data: Dict[str, Any]
) -> Any:
"""
Event that triggers on process inline_query
"""
async def on_process_chosen_inline_result(
self, chosen_inline_result: ChosenInlineResult, data: Dict[str, Any]
) -> Any:
"""
Event that triggers on process chosen_inline_result
"""
async def on_process_callback_query(
self, callback_query: CallbackQuery, data: Dict[str, Any]
) -> Any:
"""
Event that triggers on process callback_query
"""
async def on_process_shipping_query(
self, shipping_query: ShippingQuery, data: Dict[str, Any]
) -> Any:
"""
Event that triggers on process shipping_query
"""
async def on_process_pre_checkout_query(
self, pre_checkout_query: PreCheckoutQuery, data: Dict[str, Any]
) -> Any:
"""
Event that triggers on process pre_checkout_query
"""
async def on_process_poll(self, poll: Poll, data: Dict[str, Any]) -> Any:
"""
Event that triggers on process poll
"""
async def on_process_poll_answer(
self, poll_answer: PollAnswer, data: Dict[str, Any]
) -> Any:
"""
Event that triggers on process poll_answer
"""
async def on_process_error(self, exception: Exception, data: Dict[str, Any]) -> Any:
"""
Event that triggers on process error
"""
# =============================================================================================
# Event that triggers after process <event>.
# =============================================================================================
async def on_post_process_update(
self, update: Update, data: Dict[str, Any], result: Any
) -> Any:
"""
Event that triggers after processing update
"""
async def on_post_process_message(
self, message: Message, data: Dict[str, Any], result: Any
) -> Any:
"""
Event that triggers after processing message
"""
async def on_post_process_edited_message(
self, edited_message: Message, data: Dict[str, Any], result: Any
) -> Any:
"""
Event that triggers after processing edited_message
"""
async def on_post_process_channel_post(
self, channel_post: Message, data: Dict[str, Any], result: Any
) -> Any:
"""
Event that triggers after processing channel_post
"""
async def on_post_process_edited_channel_post(
self, edited_channel_post: Message, data: Dict[str, Any], result: Any
) -> Any:
"""
Event that triggers after processing edited_channel_post
"""
async def on_post_process_inline_query(
self, inline_query: InlineQuery, data: Dict[str, Any], result: Any
) -> Any:
"""
Event that triggers after processing inline_query
"""
async def on_post_process_chosen_inline_result(
self, chosen_inline_result: ChosenInlineResult, data: Dict[str, Any], result: Any
) -> Any:
"""
Event that triggers after processing chosen_inline_result
"""
async def on_post_process_callback_query(
self, callback_query: CallbackQuery, data: Dict[str, Any], result: Any
) -> Any:
"""
Event that triggers after processing callback_query
"""
async def on_post_process_shipping_query(
self, shipping_query: ShippingQuery, data: Dict[str, Any], result: Any
) -> Any:
"""
Event that triggers after processing shipping_query
"""
async def on_post_process_pre_checkout_query(
self, pre_checkout_query: PreCheckoutQuery, data: Dict[str, Any], result: Any
) -> Any:
"""
Event that triggers after processing pre_checkout_query
"""
async def on_post_process_poll(self, poll: Poll, data: Dict[str, Any], result: Any) -> Any:
"""
Event that triggers after processing poll
"""
async def on_post_process_poll_answer(
self, poll_answer: PollAnswer, data: Dict[str, Any], result: Any
) -> Any:
"""
Event that triggers after processing poll_answer
"""
async def on_post_process_error(
self, exception: Exception, data: Dict[str, Any], result: Any
) -> Any:
"""
Event that triggers after processing error
"""

71
aiogram/dispatcher/middlewares/manager.py Normal file
View file

@ -0,0 +1,71 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Dict, List
from warnings import warn
from .abstract import AbstractMiddleware
from .types import MiddlewareStep, UpdateType
if TYPE_CHECKING: # pragma: no cover
from aiogram.dispatcher.router import Router
class MiddlewareManager:
"""
Middleware manager.
"""
def __init__(self, router: Router) -> None:
self.router = router
self.middlewares: List[AbstractMiddleware] = []
def setup(self, middleware: AbstractMiddleware, _stack_level: int = 1) -> AbstractMiddleware:
"""
Setup middleware
:param middleware:
:param _stack_level:
:return:
"""
if not isinstance(middleware, AbstractMiddleware):
raise TypeError(
f"`middleware` should be instance of BaseMiddleware, not {type(middleware)}"
)
if middleware.configured:
if middleware.manager is self:
warn(
f"Middleware {middleware} is already configured for this Router "
"That's mean re-installing of this middleware has no effect.",
category=RuntimeWarning,
stacklevel=_stack_level + 1,
)
return middleware
raise ValueError(
f"Middleware is already configured for another manager {middleware.manager} "
f"in router {middleware.manager.router}!"
)
self.middlewares.append(middleware)
middleware.setup(self)
return middleware
async def trigger(
self,
step: MiddlewareStep,
event_name: str,
event: UpdateType,
data: Dict[str, Any],
result: Any = None,
reverse: bool = False,
) -> Any:
"""
Call action to middlewares with args lilt.
"""
middlewares = reversed(self.middlewares) if reverse else self.middlewares
for middleware in middlewares:
await middleware.trigger(
step=step, event_name=event_name, event=event, data=data, result=result
)
def __contains__(self, item: AbstractMiddleware) -> bool:
return item in self.middlewares

35
aiogram/dispatcher/middlewares/types.py Normal file
View file

@ -0,0 +1,35 @@
from __future__ import annotations
from enum import Enum
from typing import Union
from aiogram.api.types import (
CallbackQuery,
ChosenInlineResult,
InlineQuery,
Message,
Poll,
PollAnswer,
PreCheckoutQuery,
ShippingQuery,
Update,
)
UpdateType = Union[
CallbackQuery,
ChosenInlineResult,
InlineQuery,
Message,
Poll,
PollAnswer,
PreCheckoutQuery,
ShippingQuery,
Update,
BaseException,
]
class MiddlewareStep(Enum):
PRE_PROCESS = "pre_process"
PROCESS = "process"
POST_PROCESS = "post_process"

130
aiogram/dispatcher/router.py
View file

@ -1,13 +1,15 @@
from __future__ import annotations
import warnings
from typing import Any, Dict, List, Optional, Union
from typing import Any, Dict, Generator, List, Optional, Union
from ..api.types import Chat, Update, User
from ..api.types import Chat, TelegramObject, Update, User
from ..utils.imports import import_module
from ..utils.warnings import CodeHasNoEffect
from .event.observer import EventObserver, SkipHandler, TelegramEventObserver
from .filters import BUILTIN_FILTERS
from .middlewares.abstract import AbstractMiddleware
from .middlewares.manager import MiddlewareManager
class Router:
@ -46,6 +48,9 @@ class Router:
)
self.poll_handler = TelegramEventObserver(router=self, event_name="poll")
self.poll_answer_handler = TelegramEventObserver(router=self, event_name="poll_answer")
self.errors_handler = TelegramEventObserver(router=self, event_name="error")
self.middleware = MiddlewareManager(router=self)
self.startup = EventObserver()
self.shutdown = EventObserver()
@ -63,6 +68,7 @@ class Router:
"pre_checkout_query": self.pre_checkout_query_handler,
"poll": self.poll_handler,
"poll_answer": self.poll_answer_handler,
"error": self.errors_handler,
}
# Root handler
@ -74,6 +80,36 @@ class Router:
for builtin_filter in BUILTIN_FILTERS.get(name, ()):
observer.bind_filter(builtin_filter)
@property
def chain_head(self) -> Generator[Router, None, None]:
router: Optional[Router] = self
while router:
yield router
router = router.parent_router
@property
def chain_tail(self) -> Generator[Router, None, None]:
yield self
for router in self.sub_routers:
yield from router.chain_tail
@property
def chain(self) -> Generator[Router, None, None]:
yield from self.chain_head
tail = self.chain_tail
next(tail) # Skip self
yield from tail
def use(self, middleware: AbstractMiddleware, _stack_level: int = 1) -> AbstractMiddleware:
"""
Use middleware
:param middleware:
:param _stack_level:
:return:
"""
return self.middleware.setup(middleware, _stack_level=_stack_level + 1)
@property
def parent_router(self) -> Optional[Router]:
return self._parent_router
@ -146,11 +182,10 @@ class Router:
:param kwargs:
:return:
"""
kwargs.update(event_update=update, event_router=self)
chat: Optional[Chat] = None
from_user: Optional[User] = None
event: TelegramObject
if update.message:
update_type = "message"
from_user = update.message.from_user
@ -195,23 +230,86 @@ class Router:
update_type = "poll"
event = update.poll
else:
warnings.warn(
"Detected unknown update type.\n"
"Seems like Telegram Bot API was updated and you have "
"installed not latest version of aiogram framework",
RuntimeWarning,
)
raise SkipHandler
observer = self.observers[update_type]
if from_user:
User.set_current(from_user)
if chat:
Chat.set_current(chat)
async for result in observer.trigger(event, update=update, **kwargs):
return result
return await self.listen_update(
update_type=update_type,
update=update,
event=event,
from_user=from_user,
chat=chat,
**kwargs,
)
for router in self.sub_routers:
async for result in router.update_handler.trigger(update, **kwargs):
async def listen_update(
self,
update_type: str,
update: Update,
event: TelegramObject,
from_user: Optional[User] = None,
chat: Optional[Chat] = None,
**kwargs: Any,
) -> Any:
"""
Listen update by current and child routers
:param update_type:
:param update:
:param event:
:param from_user:
:param chat:
:param kwargs:
:return:
"""
user_token = None
if from_user:
user_token = User.set_current(from_user)
chat_token = None
if chat:
chat_token = Chat.set_current(chat)
kwargs.update(event_update=update, event_router=self)
observer = self.observers[update_type]
try:
async for result in observer.trigger(event, update=update, **kwargs):
return result
raise SkipHandler
for router in self.sub_routers:
try:
return await router.listen_update(
update_type=update_type,
update=update,
event=event,
from_user=from_user,
chat=chat,
**kwargs,
)
except SkipHandler:
continue
async def emit_startup(self, *args, **kwargs) -> None:
raise SkipHandler
except SkipHandler:
raise
except Exception as e:
async for result in self.errors_handler.trigger(e, **kwargs):
return result
raise
finally:
if user_token:
User.reset_current(user_token)
if chat_token:
Chat.reset_current(chat_token)
async def emit_startup(self, *args: Any, **kwargs: Any) -> None:
"""
Recursively call startup callbacks
@ -225,7 +323,7 @@ class Router:
for router in self.sub_routers:
await router.emit_startup(*args, **kwargs)
async def emit_shutdown(self, *args, **kwargs) -> None:
async def emit_shutdown(self, *args: Any, **kwargs: Any) -> None:
"""
Recursively call shutdown callbacks to graceful shutdown

1
aiogram/loggers.py
View file

@ -1,3 +1,4 @@
import logging
dispatcher = logging.getLogger("aiogram.dispatcher")
middlewares = logging.getLogger("aiogram.middlewares")

0
aiogram/py.typed Normal file
View file

65
aiogram/utils/helper.py
View file

@ -13,7 +13,8 @@ Example:
>>> print(MyHelper.all())
<<< ['barItem', 'bazItem', 'fooItem', 'lorem']
"""
from typing import List
import inspect
from typing import Any, Callable, Iterable, List, Optional, Union, cast
PROPS_KEYS_ATTR_NAME = "_props_keys"
@ -22,12 +23,12 @@ class Helper:
mode = ""
@classmethod
def all(cls):
def all(cls) -> List[Any]:
"""
Get all consts
:return: list
"""
result = []
result: List[Any] = []
for name in dir(cls):
if not name.isupper():
continue
@ -49,7 +50,7 @@ class HelperMode(Helper):
lowercase = "lowercase"
@classmethod
def all(cls):
def all(cls) -> List[str]:
return [
cls.SCREAMING_SNAKE_CASE,
cls.lowerCamelCase,
@ -59,7 +60,7 @@ class HelperMode(Helper):
]
@classmethod
def _screaming_snake_case(cls, text):
def _screaming_snake_case(cls, text: str) -> str:
"""
Transform text to SCREAMING_SNAKE_CASE
@ -77,7 +78,7 @@ class HelperMode(Helper):
return result
@classmethod
def _snake_case(cls, text):
def _snake_case(cls, text: str) -> str:
"""
Transform text to snake case (Based on SCREAMING_SNAKE_CASE)
@ -89,7 +90,7 @@ class HelperMode(Helper):
return cls._screaming_snake_case(text).lower()
@classmethod
def _camel_case(cls, text, first_upper=False):
def _camel_case(cls, text: str, first_upper: bool = False) -> str:
"""
Transform text to camelCase or CamelCase
@ -113,7 +114,7 @@ class HelperMode(Helper):
return result
@classmethod
def apply(cls, text, mode):
def apply(cls, text: str, mode: Union[str, Callable[[str], str]]) -> str:
"""
Apply mode for text
@ -136,7 +137,20 @@ class HelperMode(Helper):
return text
class Item:
class _BaseItem:
def __init__(self, value: Optional[str] = None):
self._value = cast(str, value)
def __set_name__(self, owner: Any, name: str) -> None:
if not name.isupper():
raise NameError("Name for helper item must be in uppercase!")
if not self._value:
if not inspect.isclass(owner) or not issubclass(owner, Helper):
raise RuntimeError("Instances of Item can be used only as Helper attributes")
self._value = HelperMode.apply(name, owner.mode)
class Item(_BaseItem):
"""
Helper item
@ -144,34 +158,24 @@ class Item:
it will be automatically generated based on a variable's name
"""
def __init__(self, value=None):
self._value = value
def __get__(self, instance, owner):
def __get__(self, instance: Any, owner: Any) -> str:
return self._value
def __set_name__(self, owner, name):
if not name.isupper():
raise NameError("Name for helper item must be in uppercase!")
if not self._value:
if hasattr(owner, "mode"):
self._value = HelperMode.apply(name, getattr(owner, "mode"))
class ListItem(Item):
class ListItem(_BaseItem):
"""
This item is always a list
You can use &, | and + operators for that.
"""
def add(self, other): # pragma: no cover
def add(self, other: "ListItem") -> "ListItem": # pragma: no cover
return self + other
def __get__(self, instance, owner):
def __get__(self, instance: Any, owner: Any) -> "ItemsList":
return ItemsList(self._value)
def __getitem__(self, item): # pragma: no cover
def __getitem__(self, item: Any) -> Any: # pragma: no cover
# Only for IDE. This method is never be called.
return self._value
@ -179,17 +183,17 @@ class ListItem(Item):
__iadd__ = __add__ = __rand__ = __and__ = __ror__ = __or__ = add
class ItemsList(list):
class ItemsList(List[str]):
"""
Patch for default list
This class provides +, &, |, +=, &=, |= operators for extending the list
"""
def __init__(self, *seq):
def __init__(self, *seq: Any):
super(ItemsList, self).__init__(map(str, seq))
def add(self, other):
def add(self, other: Iterable[str]) -> "ItemsList":
self.extend(other)
return self
@ -197,7 +201,7 @@ class ItemsList(list):
class OrderedHelperMeta(type):
def __new__(mcs, name, bases, namespace, **kwargs):
def __new__(mcs, name: Any, bases: Any, namespace: Any, **kwargs: Any) -> "OrderedHelperMeta":
cls = super().__new__(mcs, name, bases, namespace)
props_keys = []
@ -209,10 +213,11 @@ class OrderedHelperMeta(type):
setattr(cls, PROPS_KEYS_ATTR_NAME, props_keys)
return cls
# ref: https://gitter.im/python/typing?at=5da98cc5fa637359fc9cbfe1
return cast(OrderedHelperMeta, cls)
class OrderedHelper(metaclass=OrderedHelperMeta):
class OrderedHelper(Helper, metaclass=OrderedHelperMeta):
mode = ""
@classmethod

87
aiogram/utils/mixins.py
View file

@ -1,51 +1,88 @@
import contextvars
from typing import Type, TypeVar
from __future__ import annotations
__all__ = ("DataMixin", "ContextInstanceMixin")
import contextvars
from typing import Any, ClassVar, Dict, Generic, Optional, TypeVar, cast, overload
from typing_extensions import Literal
__all__ = ("ContextInstanceMixin", "DataMixin")
class DataMixin:
@property
def data(self):
data = getattr(self, "_data", None)
def data(self) -> Dict[str, Any]:
data: Optional[Dict[str, Any]] = getattr(self, "_data", None)
if data is None:
data = {}
setattr(self, "_data", data)
return data
def __getitem__(self, item):
return self.data[item]
def __getitem__(self, key: str) -> Any:
return self.data[key]
def __setitem__(self, key, value):
def __setitem__(self, key: str, value: Any) -> None:
self.data[key] = value
def __delitem__(self, key):
def __delitem__(self, key: str) -> None:
del self.data[key]
def __contains__(self, item):
return item in self.data
def __contains__(self, key: str) -> bool:
return key in self.data
def get(self, key, default=None):
def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
return self.data.get(key, default)
T = TypeVar("T")
ContextInstance = TypeVar("ContextInstance")
class ContextInstanceMixin:
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
class ContextInstanceMixin(Generic[ContextInstance]):
__context_instance: ClassVar[contextvars.ContextVar[ContextInstance]]
def __init_subclass__(cls, **kwargs: Any) -> None:
super().__init_subclass__()
cls.__context_instance = contextvars.ContextVar(f"instance_{cls.__name__}")
return cls
@overload # noqa: F811
@classmethod
def get_current(cls) -> Optional[ContextInstance]: # pragma: no cover # noqa: F811
...
@overload # noqa: F811
@classmethod
def get_current( # noqa: F811
cls, no_error: Literal[True]
) -> Optional[ContextInstance]: # pragma: no cover # noqa: F811
...
@overload # noqa: F811
@classmethod
def get_current( # noqa: F811
cls, no_error: Literal[False]
) -> ContextInstance: # pragma: no cover # noqa: F811
...
@classmethod # noqa: F811
def get_current( # noqa: F811
cls, no_error: bool = True
) -> Optional[ContextInstance]: # pragma: no cover # noqa: F811
# on mypy 0.770 I catch that contextvars.ContextVar always contextvars.ContextVar[Any]
cls.__context_instance = cast(
contextvars.ContextVar[ContextInstance], cls.__context_instance
)
try:
current: Optional[ContextInstance] = cls.__context_instance.get()
except LookupError:
if no_error:
current = None
else:
raise
return current
@classmethod
def get_current(cls: Type[T], no_error=True) -> T:
if no_error:
return cls.__context_instance.get(None)
return cls.__context_instance.get()
@classmethod
def set_current(cls: Type[T], value: T) -> contextvars.Token:
def set_current(cls, value: ContextInstance) -> contextvars.Token[ContextInstance]:
if not isinstance(value, cls):
raise TypeError(
f"Value should be instance of {cls.__name__!r} not {type(value).__name__!r}"
@ -53,5 +90,5 @@ class ContextInstanceMixin:
return cls.__context_instance.set(value)
@classmethod
def reset_current(cls: Type[T], token: contextvars.Token):
def reset_current(cls, token: contextvars.Token[ContextInstance]) -> None:
cls.__context_instance.reset(token)

2
docs/_api_version.md
View file

@ -1 +1 @@
4.6
4.7

9
docs/api/methods/add_sticker_to_set.md
View file

@ -2,7 +2,7 @@
## Description
Use this method to add a new sticker to a set created by the bot. Returns True on success.
Use this method to add a new sticker to a set created by the bot. You must use exactly one of the fields png_sticker or tgs_sticker. Animated stickers can be added to animated sticker sets and only to them. Animated sticker sets can have up to 50 stickers. Static sticker sets can have up to 120 stickers. Returns True on success.
## Arguments
@ -11,8 +11,9 @@ Use this method to add a new sticker to a set created by the bot. Returns True o
| - | - | - |
| `user_id` | `#!python3 int` | User identifier of sticker set owner |
| `name` | `#!python3 str` | Sticker set name |
| `png_sticker` | `#!python3 Union[InputFile, str]` | Png image with the sticker, must be up to 512 kilobytes in size, dimensions must not exceed 512px, and either width or height must be exactly 512px. Pass a file_id as a String to send a file that already exists on the Telegram servers, pass an HTTP URL as a String for Telegram to get a file from the Internet, or upload a new one using multipart/form-data. |
| `png_sticker` | `#!python3 Union[InputFile, str]` | PNG image with the sticker, must be up to 512 kilobytes in size, dimensions must not exceed 512px, and either width or height must be exactly 512px. Pass a file_id as a String to send a file that already exists on the Telegram servers, pass an HTTP URL as a String for Telegram to get a file from the Internet, or upload a new one using multipart/form-data. |
| `emojis` | `#!python3 str` | One or more emoji corresponding to the sticker |
| `tgs_sticker` | `#!python3 Optional[InputFile]` | Optional. TGS animation with the sticker, uploaded using multipart/form-data. See https://core.telegram.org/animated_stickers#technical-requirements for technical requirements |
| `mask_position` | `#!python3 Optional[MaskPosition]` | Optional. A JSON-serialized object for position where the mask should be placed on faces |
@ -26,8 +27,7 @@ Description: Returns True on success.
## Usage
### As bot method bot
### As bot method
```python3
result: bool = await bot.add_sticker_to_set(...)
@ -56,7 +56,6 @@ return AddStickerToSet(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#addstickertoset)

4
docs/api/methods/answer_callback_query.md
View file

@ -28,8 +28,7 @@ Description: On success, True is returned.
## Usage
### As bot method bot
### As bot method
```python3
result: bool = await bot.answer_callback_query(...)
@ -58,7 +57,6 @@ return AnswerCallbackQuery(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#answercallbackquery)

4
docs/api/methods/answer_inline_query.md
View file

@ -30,8 +30,7 @@ Description: On success, True is returned.
## Usage
### As bot method bot
### As bot method
```python3
result: bool = await bot.answer_inline_query(...)
@ -60,7 +59,6 @@ return AnswerInlineQuery(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#answerinlinequery)

4
docs/api/methods/answer_pre_checkout_query.md
View file

@ -24,8 +24,7 @@ Description: On success, True is returned.
## Usage
### As bot method bot
### As bot method
```python3
result: bool = await bot.answer_pre_checkout_query(...)
@ -54,7 +53,6 @@ return AnswerPreCheckoutQuery(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#answerprecheckoutquery)

4
docs/api/methods/answer_shipping_query.md
View file

@ -25,8 +25,7 @@ Description: On success, True is returned.
## Usage
### As bot method bot
### As bot method
```python3
result: bool = await bot.answer_shipping_query(...)
@ -55,7 +54,6 @@ return AnswerShippingQuery(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#answershippingquery)

9
docs/api/methods/create_new_sticker_set.md
View file

@ -2,7 +2,7 @@
## Description
Use this method to create new sticker set owned by a user. The bot will be able to edit the created sticker set. Returns True on success.
Use this method to create a new sticker set owned by a user. The bot will be able to edit the sticker set thus created. You must use exactly one of the fields png_sticker or tgs_sticker. Returns True on success.
## Arguments
@ -12,8 +12,9 @@ Use this method to create new sticker set owned by a user. The bot will be able
| `user_id` | `#!python3 int` | User identifier of created sticker set owner |
| `name` | `#!python3 str` | Short name of sticker set, to be used in t.me/addstickers/ URLs (e.g., animals). Can contain only english letters, digits and underscores. Must begin with a letter, can't contain consecutive underscores and must end in '_by_<bot username>'. <bot_username> is case insensitive. 1-64 characters. |
| `title` | `#!python3 str` | Sticker set title, 1-64 characters |
| `png_sticker` | `#!python3 Union[InputFile, str]` | Png image with the sticker, must be up to 512 kilobytes in size, dimensions must not exceed 512px, and either width or height must be exactly 512px. Pass a file_id as a String to send a file that already exists on the Telegram servers, pass an HTTP URL as a String for Telegram to get a file from the Internet, or upload a new one using multipart/form-data. |
| `emojis` | `#!python3 str` | One or more emoji corresponding to the sticker |
| `png_sticker` | `#!python3 Optional[Union[InputFile, str]]` | Optional. PNG image with the sticker, must be up to 512 kilobytes in size, dimensions must not exceed 512px, and either width or height must be exactly 512px. Pass a file_id as a String to send a file that already exists on the Telegram servers, pass an HTTP URL as a String for Telegram to get a file from the Internet, or upload a new one using multipart/form-data. |
| `tgs_sticker` | `#!python3 Optional[InputFile]` | Optional. TGS animation with the sticker, uploaded using multipart/form-data. See https://core.telegram.org/animated_stickers#technical-requirements for technical requirements |
| `contains_masks` | `#!python3 Optional[bool]` | Optional. Pass True, if a set of mask stickers should be created |
| `mask_position` | `#!python3 Optional[MaskPosition]` | Optional. A JSON-serialized object for position where the mask should be placed on faces |
@ -28,8 +29,7 @@ Description: Returns True on success.
## Usage
### As bot method bot
### As bot method
```python3
result: bool = await bot.create_new_sticker_set(...)
@ -58,7 +58,6 @@ return CreateNewStickerSet(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#createnewstickerset)

4
docs/api/methods/delete_chat_photo.md
View file

@ -22,8 +22,7 @@ Description: Returns True on success.
## Usage
### As bot method bot
### As bot method
```python3
result: bool = await bot.delete_chat_photo(...)
@ -52,7 +51,6 @@ return DeleteChatPhoto(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#deletechatphoto)

4
docs/api/methods/delete_chat_sticker_set.md
View file

@ -22,8 +22,7 @@ Description: Use the field can_set_sticker_set optionally returned in getChat re
## Usage
### As bot method bot
### As bot method
```python3
result: bool = await bot.delete_chat_sticker_set(...)
@ -52,7 +51,6 @@ return DeleteChatStickerSet(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#deletechatstickerset)

6
docs/api/methods/delete_message.md
View file

@ -6,6 +6,8 @@ Use this method to delete a message, including service messages, with the follow
- A message can only be deleted if it was sent less than 48 hours ago.
- A dice message in a private chat can only be deleted if it was sent more than 24 hours ago.
- Bots can delete outgoing messages in private chats, groups, and supergroups.
- Bots can delete incoming messages in private chats.
@ -37,8 +39,7 @@ Description: Returns True on success.
## Usage
### As bot method bot
### As bot method
```python3
result: bool = await bot.delete_message(...)
@ -67,7 +68,6 @@ return DeleteMessage(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#deletemessage)

4
docs/api/methods/delete_sticker_from_set.md
View file

@ -22,8 +22,7 @@ Description: Returns True on success.
## Usage
### As bot method bot
### As bot method
```python3
result: bool = await bot.delete_sticker_from_set(...)
@ -52,7 +51,6 @@ return DeleteStickerFromSet(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#deletestickerfromset)

4
docs/api/methods/delete_webhook.md
View file

@ -16,8 +16,7 @@ Description: Returns True on success.
## Usage
### As bot method bot
### As bot method
```python3
result: bool = await bot.delete_webhook(...)
@ -46,7 +45,6 @@ return DeleteWebhook(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#deletewebhook)

6
docs/api/methods/edit_message_caption.md
View file

@ -12,7 +12,7 @@ Use this method to edit captions of messages. On success, if edited message is s
| `chat_id` | `#!python3 Optional[Union[int, str]]` | Optional. Required if inline_message_id is not specified. Unique identifier for the target chat or username of the target channel (in the format @channelusername) |
| `message_id` | `#!python3 Optional[int]` | Optional. Required if inline_message_id is not specified. Identifier of the message to edit |
| `inline_message_id` | `#!python3 Optional[str]` | Optional. Required if chat_id and message_id are not specified. Identifier of the inline message |
| `caption` | `#!python3 Optional[str]` | Optional. New caption of the message |
| `caption` | `#!python3 Optional[str]` | Optional. New caption of the message, 0-1024 characters after entities parsing |
| `parse_mode` | `#!python3 Optional[str]` | Optional. Send Markdown or HTML, if you want Telegram apps to show bold, italic, fixed-width text or inline URLs in the media caption. |
| `reply_markup` | `#!python3 Optional[InlineKeyboardMarkup]` | Optional. A JSON-serialized object for an inline keyboard. |
@ -27,8 +27,7 @@ Description: On success, if edited message is sent by the bot, the edited Messag
## Usage
### As bot method bot
### As bot method
```python3
result: Union[Message, bool] = await bot.edit_message_caption(...)
@ -57,7 +56,6 @@ return EditMessageCaption(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#editmessagecaption)

4
docs/api/methods/edit_message_live_location.md
View file

@ -27,8 +27,7 @@ Description: On success, if the edited message was sent by the bot, the edited M
## Usage
### As bot method bot
### As bot method
```python3
result: Union[Message, bool] = await bot.edit_message_live_location(...)
@ -57,7 +56,6 @@ return EditMessageLiveLocation(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#editmessagelivelocation)

4
docs/api/methods/edit_message_media.md
View file

@ -26,8 +26,7 @@ Description: On success, if the edited message was sent by the bot, the edited M
## Usage
### As bot method bot
### As bot method
```python3
result: Union[Message, bool] = await bot.edit_message_media(...)
@ -56,7 +55,6 @@ return EditMessageMedia(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#editmessagemedia)

4
docs/api/methods/edit_message_reply_markup.md
View file

@ -25,8 +25,7 @@ Description: On success, if edited message is sent by the bot, the edited Messag
## Usage
### As bot method bot
### As bot method
```python3
result: Union[Message, bool] = await bot.edit_message_reply_markup(...)
@ -55,7 +54,6 @@ return EditMessageReplyMarkup(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#editmessagereplymarkup)

6
docs/api/methods/edit_message_text.md
View file

@ -9,7 +9,7 @@ Use this method to edit text and game messages. On success, if edited message is
| Name | Type | Description |
| - | - | - |
| `text` | `#!python3 str` | New text of the message |
| `text` | `#!python3 str` | New text of the message, 1-4096 characters after entities parsing |
| `chat_id` | `#!python3 Optional[Union[int, str]]` | Optional. Required if inline_message_id is not specified. Unique identifier for the target chat or username of the target channel (in the format @channelusername) |
| `message_id` | `#!python3 Optional[int]` | Optional. Required if inline_message_id is not specified. Identifier of the message to edit |
| `inline_message_id` | `#!python3 Optional[str]` | Optional. Required if chat_id and message_id are not specified. Identifier of the inline message |
@ -28,8 +28,7 @@ Description: On success, if edited message is sent by the bot, the edited Messag
## Usage
### As bot method bot
### As bot method
```python3
result: Union[Message, bool] = await bot.edit_message_text(...)
@ -58,7 +57,6 @@ return EditMessageText(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#editmessagetext)

4
docs/api/methods/export_chat_invite_link.md
View file

@ -24,8 +24,7 @@ Description: Returns the new invite link as String on success.
## Usage
### As bot method bot
### As bot method
```python3
result: str = await bot.export_chat_invite_link(...)
@ -54,7 +53,6 @@ return ExportChatInviteLink(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#exportchatinvitelink)

4
docs/api/methods/forward_message.md
View file

@ -25,8 +25,7 @@ Description: On success, the sent Message is returned.
## Usage
### As bot method bot
### As bot method
```python3
result: Message = await bot.forward_message(...)
@ -55,7 +54,6 @@ return ForwardMessage(...)
```
## Related pages:
- [Official documentation](https://core.telegram.org/bots/api#forwardmessage)

3
docs/api/methods/get_chat.md
View file

@ -22,8 +22,7 @@ Description: Returns a Chat object on success.
## Usage
### As bot method bot
### As bot method
```python3
result: Chat = await bot.get_chat(...)

3
docs/api/methods/get_chat_administrators.md
View file

@ -22,8 +22,7 @@ Description: On success, returns an Array of ChatMember objects that contains in
## Usage
### As bot method bot
### As bot method
```python3
result: List[ChatMember] = await bot.get_chat_administrators(...)

Some files were not shown because too many files have changed in this diff Show more