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 5

PoC: Bot instance inside method shortcuts using pydantic Validation Context (#1210)

Browse source 
* PoC: Mount objects to the Bot instance, bind shortcuts to configured instance

* Fixe docstring of the bind method

* Pass Bot instance explicitly to the URLInputFile

* Added tests

* Added changelog

* Refactor aiogram client and update tests

Refactored base.py to improve code readability by separating response_type operation from model_validate(). Also, adjusted the parameters in URLInputFile() within test_input_file.py for better test coverage. Updated input_file.py to streamline read method and avoid unnecessary instantiation of Bot class. Lastly, adjusted typing in methods/base.py to enhance code clarity.

* Update changelog
This commit is contained in:
Alex Root Junior 2023-07-11 23:17:26 +03:00 • committed by GitHub
parent c39a803747
commit a7b92bb050
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 228 additions and 111 deletions
Download patch file Download diff file Expand all files Collapse all files

6
CHANGES/1210.misc.rst Normal file
View file

@ -0,0 +1,6 @@
Replaced ContextVar's with a new feature called `Validation Context <https://docs.pydantic.dev/latest/usage/validators/#validation-context>`_
in Pydantic to improve the clarity, usability, and versatility of handling the Bot instance within method shortcuts.
.. danger::
**Breaking**: The 'bot' argument now is required in `URLInputFile`

27
aiogram/client/context_controller.py Normal file
View file

@ -0,0 +1,27 @@
from typing import TYPE_CHECKING, Any, Optional
from pydantic import BaseModel, PrivateAttr
from typing_extensions import Self
if TYPE_CHECKING:
from aiogram.client.bot import Bot
class BotContextController(BaseModel):
_bot: Optional["Bot"] = PrivateAttr()
def model_post_init(self, __context: Any) -> None:
if not __context:
self._bot = None
else:
self._bot = __context.get("bot")
def as_(self, bot: Optional["Bot"]) -> Self:
"""
Bind object to a bot instance.
:param bot: Bot instance
:return: self
"""
self._bot = bot
return self

4
aiogram/client/session/aiohttp.py
View file

@ -167,7 +167,9 @@ class AiohttpSession(BaseSession):
raise TelegramNetworkError(method=method, message="Request timeout error") raise TelegramNetworkError(method=method, message="Request timeout error")
except ClientError as e: except ClientError as e:
raise TelegramNetworkError(method=method, message=f"{type(e).__name__}: {e}") raise TelegramNetworkError(method=method, message=f"{type(e).__name__}: {e}")
response = self.check_response(method=method, status_code=resp.status, content=raw_result) response = self.check_response(
bot=bot, method=method, status_code=resp.status, content=raw_result
)
return cast(TelegramType, response.result) return cast(TelegramType, response.result)
async def stream_content( async def stream_content(

5
aiogram/client/session/base.py
View file

@ -75,7 +75,7 @@ class BaseSession(abc.ABC):
self.middleware = RequestMiddlewareManager() self.middleware = RequestMiddlewareManager()
def check_response( def check_response(
self, method: TelegramMethod[TelegramType], status_code: int, content: str self, bot: Bot, method: TelegramMethod[TelegramType], status_code: int, content: str
) -> Response[TelegramType]: ) -> Response[TelegramType]:
""" """
Check response status Check response status
@ -89,7 +89,8 @@ class BaseSession(abc.ABC):
raise ClientDecodeError("Failed to decode object", e, content) raise ClientDecodeError("Failed to decode object", e, content)
try: try:
response = method.build_response(json_data) response_type = Response[method.__returning__] # type: ignore
response = response_type.model_validate(json_data, context={"bot": bot})
except ValidationError as e: except ValidationError as e:
raise ClientDecodeError("Failed to deserialize object", e, json_data) raise ClientDecodeError("Failed to deserialize object", e, json_data)

50
aiogram/methods/base.py
View file

@ -1,11 +1,22 @@
from __future__ import annotations from __future__ import annotations
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any, Dict, Generator, Generic, Optional, TypeVar from typing import (
TYPE_CHECKING,
Any,
ClassVar,
Dict,
Generator,
Generic,
Optional,
TypeVar,
)
from pydantic import BaseModel, ConfigDict from pydantic import BaseModel, ConfigDict
from pydantic.functional_validators import model_validator from pydantic.functional_validators import model_validator
from aiogram.client.context_controller import BotContextController
from ..types import InputFile, ResponseParameters from ..types import InputFile, ResponseParameters
from ..types.base import UNSET_TYPE from ..types.base import UNSET_TYPE
@ -32,7 +43,7 @@ class Response(BaseModel, Generic[TelegramType]):
parameters: Optional[ResponseParameters] = None parameters: Optional[ResponseParameters] = None
class TelegramMethod(BaseModel, Generic[TelegramType], ABC): class TelegramMethod(BotContextController, BaseModel, Generic[TelegramType], ABC):
model_config = ConfigDict( model_config = ConfigDict(
extra="allow", extra="allow",
populate_by_name=True, populate_by_name=True,
@ -40,6 +51,7 @@ class TelegramMethod(BaseModel, Generic[TelegramType], ABC):
) )
@model_validator(mode="before") @model_validator(mode="before")
@classmethod
def remove_unset(cls, values: Dict[str, Any]) -> Dict[str, Any]: def remove_unset(cls, values: Dict[str, Any]) -> Dict[str, Any]:
""" """
Remove UNSET before fields validation. Remove UNSET before fields validation.
@ -51,25 +63,31 @@ class TelegramMethod(BaseModel, Generic[TelegramType], ABC):
""" """
return {k: v for k, v in values.items() if not isinstance(v, UNSET_TYPE)} return {k: v for k, v in values.items() if not isinstance(v, UNSET_TYPE)}
@property if TYPE_CHECKING:
@abstractmethod __returning__: ClassVar[type]
def __returning__(self) -> type: # pragma: no cover __api_method__: ClassVar[str]
pass else:
@property @property
@abstractmethod @abstractmethod
def __api_method__(self) -> str: def __returning__(self) -> type:
pass pass
def build_response(self, data: Dict[str, Any]) -> Response[TelegramType]: @property
# noinspection PyTypeChecker @abstractmethod
return Response[self.__returning__](**data) # type: ignore def __api_method__(self) -> str:
pass
async def emit(self, bot: Bot) -> TelegramType: async def emit(self, bot: Bot) -> TelegramType:
return await bot(self) return await bot(self)
def __await__(self) -> Generator[Any, None, TelegramType]: def __await__(self) -> Generator[Any, None, TelegramType]:
from aiogram.client.bot import Bot bot = self._bot
if not bot:
bot = Bot.get_current(no_error=False) raise RuntimeError(
"This method is not mounted to a any bot instance, please call it explicilty "
"with bot instance `await bot(method)`\n"
"or mount method to a bot instance `method.as_(bot)` "
"and then call it `await method()`"
)
return self.emit(bot).__await__() return self.emit(bot).__await__()

4
aiogram/types/base.py
View file

@ -3,10 +3,10 @@ from unittest.mock import sentinel
from pydantic import BaseModel, ConfigDict from pydantic import BaseModel, ConfigDict
from aiogram.utils.mixins import ContextInstanceMixin from aiogram.client.context_controller import BotContextController
class TelegramObject(ContextInstanceMixin["TelegramObject"], BaseModel): class TelegramObject(BotContextController, BaseModel):
model_config = ConfigDict( model_config = ConfigDict(
use_enum_values=True, use_enum_values=True,
extra="allow", extra="allow",

2
aiogram/types/callback_query.py
View file

@ -74,4 +74,4 @@ class CallbackQuery(TelegramObject):
url=url, url=url,
cache_time=cache_time, cache_time=cache_time,
**kwargs, **kwargs,
) ).as_(self._bot)

54
aiogram/types/chat.py
View file

@ -164,7 +164,7 @@ class Chat(TelegramObject):
chat_id=self.id, chat_id=self.id,
sender_chat_id=sender_chat_id, sender_chat_id=sender_chat_id,
**kwargs, **kwargs,
) ).as_(self._bot)
def unban_sender_chat( def unban_sender_chat(
self, self,
@ -193,7 +193,7 @@ class Chat(TelegramObject):
chat_id=self.id, chat_id=self.id,
sender_chat_id=sender_chat_id, sender_chat_id=sender_chat_id,
**kwargs, **kwargs,
) ).as_(self._bot)
def get_administrators( def get_administrators(
self, self,
@ -219,7 +219,7 @@ class Chat(TelegramObject):
return GetChatAdministrators( return GetChatAdministrators(
chat_id=self.id, chat_id=self.id,
**kwargs, **kwargs,
) ).as_(self._bot)
def delete_message( def delete_message(
self, self,
@ -266,7 +266,7 @@ class Chat(TelegramObject):
chat_id=self.id, chat_id=self.id,
message_id=message_id, message_id=message_id,
**kwargs, **kwargs,
) ).as_(self._bot)
def revoke_invite_link( def revoke_invite_link(
self, self,
@ -295,7 +295,7 @@ class Chat(TelegramObject):
chat_id=self.id, chat_id=self.id,
invite_link=invite_link, invite_link=invite_link,
**kwargs, **kwargs,
) ).as_(self._bot)
def edit_invite_link( def edit_invite_link(
self, self,
@ -336,7 +336,7 @@ class Chat(TelegramObject):
member_limit=member_limit, member_limit=member_limit,
creates_join_request=creates_join_request, creates_join_request=creates_join_request,
**kwargs, **kwargs,
) ).as_(self._bot)
def create_invite_link( def create_invite_link(
self, self,
@ -374,7 +374,7 @@ class Chat(TelegramObject):
member_limit=member_limit, member_limit=member_limit,
creates_join_request=creates_join_request, creates_join_request=creates_join_request,
**kwargs, **kwargs,
) ).as_(self._bot)
def export_invite_link( def export_invite_link(
self, self,
@ -402,7 +402,7 @@ class Chat(TelegramObject):
return ExportChatInviteLink( return ExportChatInviteLink(
chat_id=self.id, chat_id=self.id,
**kwargs, **kwargs,
) ).as_(self._bot)
def do( def do(
self, self,
@ -438,7 +438,7 @@ class Chat(TelegramObject):
action=action, action=action,
message_thread_id=message_thread_id, message_thread_id=message_thread_id,
**kwargs, **kwargs,
) ).as_(self._bot)
def delete_sticker_set( def delete_sticker_set(
self, self,
@ -464,7 +464,7 @@ class Chat(TelegramObject):
return DeleteChatStickerSet( return DeleteChatStickerSet(
chat_id=self.id, chat_id=self.id,
**kwargs, **kwargs,
) ).as_(self._bot)
def set_sticker_set( def set_sticker_set(
self, self,
@ -493,7 +493,7 @@ class Chat(TelegramObject):
chat_id=self.id, chat_id=self.id,
sticker_set_name=sticker_set_name, sticker_set_name=sticker_set_name,
**kwargs, **kwargs,
) ).as_(self._bot)
def get_member( def get_member(
self, self,
@ -522,7 +522,7 @@ class Chat(TelegramObject):
chat_id=self.id, chat_id=self.id,
user_id=user_id, user_id=user_id,
**kwargs, **kwargs,
) ).as_(self._bot)
def get_member_count( def get_member_count(
self, self,
@ -548,7 +548,7 @@ class Chat(TelegramObject):
return GetChatMemberCount( return GetChatMemberCount(
chat_id=self.id, chat_id=self.id,
**kwargs, **kwargs,
) ).as_(self._bot)
def leave( def leave(
self, self,
@ -574,7 +574,7 @@ class Chat(TelegramObject):
return LeaveChat( return LeaveChat(
chat_id=self.id, chat_id=self.id,
**kwargs, **kwargs,
) ).as_(self._bot)
def unpin_all_messages( def unpin_all_messages(
self, self,
@ -600,7 +600,7 @@ class Chat(TelegramObject):
return UnpinAllChatMessages( return UnpinAllChatMessages(
chat_id=self.id, chat_id=self.id,
**kwargs, **kwargs,
) ).as_(self._bot)
def unpin_message( def unpin_message(
self, self,
@ -629,7 +629,7 @@ class Chat(TelegramObject):
chat_id=self.id, chat_id=self.id,
message_id=message_id, message_id=message_id,
**kwargs, **kwargs,
) ).as_(self._bot)
def pin_message( def pin_message(
self, self,
@ -661,7 +661,7 @@ class Chat(TelegramObject):
message_id=message_id, message_id=message_id,
disable_notification=disable_notification, disable_notification=disable_notification,
**kwargs, **kwargs,
) ).as_(self._bot)
def set_administrator_custom_title( def set_administrator_custom_title(
self, self,
@ -693,7 +693,7 @@ class Chat(TelegramObject):
user_id=user_id, user_id=user_id,
custom_title=custom_title, custom_title=custom_title,
**kwargs, **kwargs,
) ).as_(self._bot)
def set_permissions( def set_permissions(
self, self,
@ -725,7 +725,7 @@ class Chat(TelegramObject):
permissions=permissions, permissions=permissions,
use_independent_chat_permissions=use_independent_chat_permissions, use_independent_chat_permissions=use_independent_chat_permissions,
**kwargs, **kwargs,
) ).as_(self._bot)
def promote( def promote(
self, self,
@ -790,7 +790,7 @@ class Chat(TelegramObject):
can_pin_messages=can_pin_messages, can_pin_messages=can_pin_messages,
can_manage_topics=can_manage_topics, can_manage_topics=can_manage_topics,
**kwargs, **kwargs,
) ).as_(self._bot)
def restrict( def restrict(
self, self,
@ -828,7 +828,7 @@ class Chat(TelegramObject):
use_independent_chat_permissions=use_independent_chat_permissions, use_independent_chat_permissions=use_independent_chat_permissions,
until_date=until_date, until_date=until_date,
**kwargs, **kwargs,
) ).as_(self._bot)
def unban( def unban(
self, self,
@ -860,7 +860,7 @@ class Chat(TelegramObject):
user_id=user_id, user_id=user_id,
only_if_banned=only_if_banned, only_if_banned=only_if_banned,
**kwargs, **kwargs,
) ).as_(self._bot)
def ban( def ban(
self, self,
@ -895,7 +895,7 @@ class Chat(TelegramObject):
until_date=until_date, until_date=until_date,
revoke_messages=revoke_messages, revoke_messages=revoke_messages,
**kwargs, **kwargs,
) ).as_(self._bot)
def set_description( def set_description(
self, self,
@ -924,7 +924,7 @@ class Chat(TelegramObject):
chat_id=self.id, chat_id=self.id,
description=description, description=description,
**kwargs, **kwargs,
) ).as_(self._bot)
def set_title( def set_title(
self, self,
@ -953,7 +953,7 @@ class Chat(TelegramObject):
chat_id=self.id, chat_id=self.id,
title=title, title=title,
**kwargs, **kwargs,
) ).as_(self._bot)
def delete_photo( def delete_photo(
self, self,
@ -979,7 +979,7 @@ class Chat(TelegramObject):
return DeleteChatPhoto( return DeleteChatPhoto(
chat_id=self.id, chat_id=self.id,
**kwargs, **kwargs,
) ).as_(self._bot)
def set_photo( def set_photo(
self, self,
@ -1008,4 +1008,4 @@ class Chat(TelegramObject):
chat_id=self.id, chat_id=self.id,
photo=photo, photo=photo,
**kwargs, **kwargs,
) ).as_(self._bot)

4
aiogram/types/chat_join_request.py
View file

@ -62,7 +62,7 @@ class ChatJoinRequest(TelegramObject):
chat_id=self.chat.id, chat_id=self.chat.id,
user_id=self.from_user.id, user_id=self.from_user.id,
**kwargs, **kwargs,
) ).as_(self._bot)
def decline( def decline(
self, self,
@ -90,4 +90,4 @@ class ChatJoinRequest(TelegramObject):
chat_id=self.chat.id, chat_id=self.chat.id,
user_id=self.from_user.id, user_id=self.from_user.id,
**kwargs, **kwargs,
) ).as_(self._bot)

2
aiogram/types/inline_query.py
View file

@ -81,4 +81,4 @@ class InlineQuery(TelegramObject):
switch_pm_parameter=switch_pm_parameter, switch_pm_parameter=switch_pm_parameter,
switch_pm_text=switch_pm_text, switch_pm_text=switch_pm_text,
**kwargs, **kwargs,
) ).as_(self._bot)

23
aiogram/types/input_file.py
View file

@ -4,10 +4,21 @@ import io
import os import os
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from pathlib import Path from pathlib import Path
from typing import Any, AsyncGenerator, AsyncIterator, Dict, Iterator, Optional, Union from typing import (
TYPE_CHECKING,
Any,
AsyncGenerator,
AsyncIterator,
Dict,
Optional,
Union,
)
import aiofiles import aiofiles
if TYPE_CHECKING:
from aiogram.client.bot import Bot
DEFAULT_CHUNK_SIZE = 64 * 1024 # 64 kb DEFAULT_CHUNK_SIZE = 64 * 1024 # 64 kb
@ -110,6 +121,7 @@ class URLInputFile(InputFile):
def __init__( def __init__(
self, self,
url: str, url: str,
bot: "Bot",
headers: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, Any]] = None,
filename: Optional[str] = None, filename: Optional[str] = None,
chunk_size: int = DEFAULT_CHUNK_SIZE, chunk_size: int = DEFAULT_CHUNK_SIZE,
@ -122,6 +134,9 @@ class URLInputFile(InputFile):
:param headers: HTTP Headers :param headers: HTTP Headers
:param filename: Filename to be propagated to telegram. :param filename: Filename to be propagated to telegram.
:param chunk_size: Uploading chunk size :param chunk_size: Uploading chunk size
:param timeout: Timeout for downloading
:param bot: Bot instance to use HTTP session from.
If not specified, will be used current bot from context.
""" """
super().__init__(filename=filename, chunk_size=chunk_size) super().__init__(filename=filename, chunk_size=chunk_size)
if headers is None: if headers is None:
@ -130,12 +145,10 @@ class URLInputFile(InputFile):
self.url = url self.url = url
self.headers = headers self.headers = headers
self.timeout = timeout self.timeout = timeout
self.bot = bot
async def read(self, chunk_size: int) -> AsyncGenerator[bytes, None]: async def read(self, chunk_size: int) -> AsyncGenerator[bytes, None]:
from aiogram.client.bot import Bot stream = self.bot.session.stream_content(
bot = Bot.get_current(no_error=False)
stream = bot.session.stream_content(
url=self.url, url=self.url,
headers=self.headers, headers=self.headers,
timeout=self.timeout, timeout=self.timeout,

90
aiogram/types/message.py
View file

@ -418,7 +418,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_animation( def answer_animation(
self, self,
@ -490,7 +490,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_audio( def reply_audio(
self, self,
@ -559,7 +559,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_audio( def answer_audio(
self, self,
@ -629,7 +629,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_contact( def reply_contact(
self, self,
@ -685,7 +685,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_contact( def answer_contact(
self, self,
@ -742,7 +742,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_document( def reply_document(
self, self,
@ -804,7 +804,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_document( def answer_document(
self, self,
@ -867,7 +867,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_game( def reply_game(
self, self,
@ -912,7 +912,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_game( def answer_game(
self, self,
@ -958,7 +958,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_invoice( def reply_invoice(
self, self,
@ -1063,7 +1063,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_invoice( def answer_invoice(
self, self,
@ -1169,7 +1169,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_location( def reply_location(
self, self,
@ -1231,7 +1231,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_location( def answer_location(
self, self,
@ -1294,7 +1294,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_media_group( def reply_media_group(
self, self,
@ -1336,7 +1336,7 @@ class Message(TelegramObject):
protect_content=protect_content, protect_content=protect_content,
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_media_group( def answer_media_group(
self, self,
@ -1379,7 +1379,7 @@ class Message(TelegramObject):
reply_to_message_id=reply_to_message_id, reply_to_message_id=reply_to_message_id,
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply( def reply(
self, self,
@ -1435,7 +1435,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer( def answer(
self, self,
@ -1492,7 +1492,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_photo( def reply_photo(
self, self,
@ -1551,7 +1551,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_photo( def answer_photo(
self, self,
@ -1611,7 +1611,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_poll( def reply_poll(
self, self,
@ -1691,7 +1691,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_poll( def answer_poll(
self, self,
@ -1772,7 +1772,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_dice( def reply_dice(
self, self,
@ -1819,7 +1819,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_dice( def answer_dice(
self, self,
@ -1867,7 +1867,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_sticker( def reply_sticker(
self, self,
@ -1917,7 +1917,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_sticker( def answer_sticker(
self, self,
@ -1968,7 +1968,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_venue( def reply_venue(
self, self,
@ -2036,7 +2036,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_venue( def answer_venue(
self, self,
@ -2105,7 +2105,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_video( def reply_video(
self, self,
@ -2179,7 +2179,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_video( def answer_video(
self, self,
@ -2254,7 +2254,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_video_note( def reply_video_note(
self, self,
@ -2310,7 +2310,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_video_note( def answer_video_note(
self, self,
@ -2367,7 +2367,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def reply_voice( def reply_voice(
self, self,
@ -2426,7 +2426,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def answer_voice( def answer_voice(
self, self,
@ -2486,7 +2486,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def send_copy( # noqa: C901 def send_copy( # noqa: C901
self: Message, self: Message,
@ -2684,7 +2684,7 @@ class Message(TelegramObject):
allow_sending_without_reply=allow_sending_without_reply, allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def edit_text( def edit_text(
self, self,
@ -2730,7 +2730,7 @@ class Message(TelegramObject):
disable_web_page_preview=disable_web_page_preview, disable_web_page_preview=disable_web_page_preview,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def forward( def forward(
self, self,
@ -2770,7 +2770,7 @@ class Message(TelegramObject):
disable_notification=disable_notification, disable_notification=disable_notification,
protect_content=protect_content, protect_content=protect_content,
**kwargs, **kwargs,
) ).as_(self._bot)
def edit_media( def edit_media(
self, self,
@ -2807,7 +2807,7 @@ class Message(TelegramObject):
inline_message_id=inline_message_id, inline_message_id=inline_message_id,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def edit_reply_markup( def edit_reply_markup(
self, self,
@ -2841,7 +2841,7 @@ class Message(TelegramObject):
inline_message_id=inline_message_id, inline_message_id=inline_message_id,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def delete_reply_markup(self) -> EditMessageReplyMarkup: def delete_reply_markup(self) -> EditMessageReplyMarkup:
return self.edit_reply_markup(reply_markup=None) return self.edit_reply_markup(reply_markup=None)
@ -2893,7 +2893,7 @@ class Message(TelegramObject):
proximity_alert_radius=proximity_alert_radius, proximity_alert_radius=proximity_alert_radius,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def stop_live_location( def stop_live_location(
self, self,
@ -2927,7 +2927,7 @@ class Message(TelegramObject):
inline_message_id=inline_message_id, inline_message_id=inline_message_id,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def edit_caption( def edit_caption(
self, self,
@ -2970,7 +2970,7 @@ class Message(TelegramObject):
caption_entities=caption_entities, caption_entities=caption_entities,
reply_markup=reply_markup, reply_markup=reply_markup,
**kwargs, **kwargs,
) ).as_(self._bot)
def delete( def delete(
self, self,
@ -3016,7 +3016,7 @@ class Message(TelegramObject):
chat_id=self.chat.id, chat_id=self.chat.id,
message_id=self.message_id, message_id=self.message_id,
**kwargs, **kwargs,
) ).as_(self._bot)
def pin( def pin(
self, self,
@ -3047,7 +3047,7 @@ class Message(TelegramObject):
message_id=self.message_id, message_id=self.message_id,
disable_notification=disable_notification, disable_notification=disable_notification,
**kwargs, **kwargs,
) ).as_(self._bot)
def unpin( def unpin(
self, self,
@ -3075,7 +3075,7 @@ class Message(TelegramObject):
chat_id=self.chat.id, chat_id=self.chat.id,
message_id=self.message_id, message_id=self.message_id,
**kwargs, **kwargs,
) ).as_(self._bot)
def get_url(self, force_private: bool = False) -> Optional[str]: def get_url(self, force_private: bool = False) -> Optional[str]:
""" """

4
aiogram/types/sticker.py
View file

@ -76,7 +76,7 @@ class Sticker(TelegramObject):
sticker=self.file_id, sticker=self.file_id,
position=position, position=position,
**kwargs, **kwargs,
) ).as_(self._bot)
def delete_from_set( def delete_from_set(
self, self,
@ -102,4 +102,4 @@ class Sticker(TelegramObject):
return DeleteStickerFromSet( return DeleteStickerFromSet(
sticker=self.file_id, sticker=self.file_id,
**kwargs, **kwargs,
) ).as_(self._bot)

2
aiogram/types/user.py
View file

@ -90,4 +90,4 @@ class User(TelegramObject):
offset=offset, offset=offset,
limit=limit, limit=limit,
**kwargs, **kwargs,
) ).as_(self._bot)

2
tests/mocked_bot.py
View file

@ -35,7 +35,7 @@ class MockedSession(BaseSession):
self.requests.append(method) self.requests.append(method)
response: Response[TelegramType] = self.responses.pop() response: Response[TelegramType] = self.responses.pop()
self.check_response( self.check_response(
method=method, status_code=response.error_code, content=response.json() bot=bot, method=method, status_code=response.error_code, content=response.json()
) )
return response.result # type: ignore return response.result # type: ignore

36
tests/test_api/test_client/test_context_controller.py Normal file
View file

@ -0,0 +1,36 @@
from aiogram.client.context_controller import BotContextController
from tests.mocked_bot import MockedBot
class MyModel(BotContextController):
id: int
class TestBotContextController:
def test_via_model_validate(self, bot: MockedBot):
my_model = MyModel.model_validate({"id": 1}, context={"bot": bot})
assert my_model.id == 1
assert my_model._bot == bot
def test_via_model_validate_none(self):
my_model = MyModel.model_validate({"id": 1}, context={})
assert my_model.id == 1
assert my_model._bot is None
def test_as(self, bot: MockedBot):
my_model = MyModel(id=1).as_(bot)
assert my_model.id == 1
assert my_model._bot == bot
def test_as_none(self):
my_model = MyModel(id=1).as_(None)
assert my_model.id == 1
assert my_model._bot is None
def test_replacement(self, bot: MockedBot):
my_model = MyModel(id=1).as_(bot)
assert my_model.id == 1
assert my_model._bot == bot
my_model = my_model.as_(None)
assert my_model.id == 1
assert my_model._bot is None

7
tests/test_api/test_client/test_session/test_base_session.py
View file

@ -170,9 +170,11 @@ class TestBaseSession:
) )
def test_check_response(self, status_code, content, error): def test_check_response(self, status_code, content, error):
session = CustomSession() session = CustomSession()
bot = MockedBot()
method = DeleteMessage(chat_id=42, message_id=42) method = DeleteMessage(chat_id=42, message_id=42)
if error is None: if error is None:
session.check_response( session.check_response(
bot=bot,
method=method, method=method,
status_code=status_code, status_code=status_code,
content=content, content=content,
@ -180,6 +182,7 @@ class TestBaseSession:
else: else:
with pytest.raises(error) as exc_info: with pytest.raises(error) as exc_info:
session.check_response( session.check_response(
bot=bot,
method=method, method=method,
status_code=status_code, status_code=status_code,
content=content, content=content,
@ -191,10 +194,12 @@ class TestBaseSession:
def test_check_response_json_decode_error(self): def test_check_response_json_decode_error(self):
session = CustomSession() session = CustomSession()
bot = MockedBot()
method = DeleteMessage(chat_id=42, message_id=42) method = DeleteMessage(chat_id=42, message_id=42)
with pytest.raises(ClientDecodeError, match="JSONDecodeError"): with pytest.raises(ClientDecodeError, match="JSONDecodeError"):
session.check_response( session.check_response(
bot=bot,
method=method, method=method,
status_code=200, status_code=200,
content="is not a JSON object", content="is not a JSON object",
@ -202,10 +207,12 @@ class TestBaseSession:
def test_check_response_validation_error(self): def test_check_response_validation_error(self):
session = CustomSession() session = CustomSession()
bot = MockedBot()
method = DeleteMessage(chat_id=42, message_id=42) method = DeleteMessage(chat_id=42, message_id=42)
with pytest.raises(ClientDecodeError, match="ValidationError"): with pytest.raises(ClientDecodeError, match="ValidationError"):
session.check_response( session.check_response(
bot=bot,
method=method, method=method,
status_code=200, status_code=200,
content='{"ok": "test"}', content='{"ok": "test"}',

10
tests/test_api/test_methods/test_base.py
View file

@ -22,6 +22,14 @@ class TestTelegramMethodRemoveUnset:
class TestTelegramMethodCall: class TestTelegramMethodCall:
async def test_async_emit_unsuccessful(self, bot: MockedBot):
with pytest.raises(
RuntimeError,
match="This method is not mounted to a any bot instance.+",
):
await GetMe()
async def test_async_emit(self, bot: MockedBot): async def test_async_emit(self, bot: MockedBot):
bot.add_result_for(GetMe, ok=True, result=User(id=42, is_bot=True, first_name="Test")) bot.add_result_for(GetMe, ok=True, result=User(id=42, is_bot=True, first_name="Test"))
assert isinstance(await GetMe(), User) method = GetMe().as_(bot)
assert isinstance(await method, User)

7
tests/test_api/test_types/test_input_file.py
View file

@ -4,6 +4,7 @@ from aresponses import ResponsesMockServer
from aiogram import Bot from aiogram import Bot
from aiogram.types import BufferedInputFile, FSInputFile, InputFile, URLInputFile from aiogram.types import BufferedInputFile, FSInputFile, InputFile, URLInputFile
from tests.mocked_bot import MockedBot
class TestInputFile: class TestInputFile:
@ -72,10 +73,8 @@ class TestInputFile:
aresponses.add( aresponses.add(
aresponses.ANY, aresponses.ANY, "get", aresponses.Response(status=200, body=b"\f" * 10) aresponses.ANY, aresponses.ANY, "get", aresponses.Response(status=200, body=b"\f" * 10)
) )
bot = Bot(token="42:TEST")
Bot.set_current(Bot("42:TEST")) file = URLInputFile("https://test.org/", bot, chunk_size=1)
file = URLInputFile("https://test.org/", chunk_size=1)
size = 0 size = 0
async for chunk in file: async for chunk in file: