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

Drop py3.9 and pypy3.9

Browse source 
Add pypy3.11 (testing) into `tests.yml`

Remove py3.9 from matrix in `tests.yml`

Refactor not auto-gen code to be compatible with py3.10+, droping ugly 3.9 annotation.

Replace some `from typing` imports to `from collections.abc`, due to deprecation

Add `from __future__ import annotations` and `if TYPE_CHECKING:` where possible

Add some `noqa` to calm down Ruff in some places, if Ruff will be used as default linting+formatting tool in future

Replace some relative imports to absolute

Sort `__all__` tuples in `__init__.py` and some other `.py` files

Sort `__slots__` tuples in classes

Split raises into `msg` and `raise` (`EM101`, `EM102`) to not duplicate error message in the traceback

Add `Self` from `typing_extenstion` where possible

Resolve typing problem in `aiogram/filters/command.py:18`

Concatenate nested `if` statements

Convert `HandlerContainer` into a dataclass in `aiogram/fsm/scene.py`

Bump tests docker-compose.yml `redis:6-alpine` -> `redis:8-alpine`

Bump tests docker-compose.yml `mongo:7.0.6` -> `mongo:8.0.14`

Bump pre-commit-config `black==24.4.2` -> `black==25.9.0`

Bump pre-commit-config `ruff==0.5.1` -> `ruff==0.13.3`

Update Makefile lint for ruff to show fixes

Add `make outdated` into Makefile

Use `pathlib` instead of `os.path`

Bump `redis[hiredis]>=5.0.1,<5.3.0` -> `redis[hiredis]>=6.2.0,<7`

Bump `cryptography>=43.0.0` -> `cryptography>=46.0.0` due to security reasons

Bump `pytz~=2023.3` -> `pytz~=2025.2`

Bump `pycryptodomex~=3.19.0` -> `pycryptodomex~=3.23.0` due to security reasons

Bump linting and formatting tools
This commit is contained in:
andrew000 2025-10-03 05:13:09 +00:00
parent ab32296d07
commit e94e33c496
No known key found for this signature in database
GPG key ID: F22DA743FBEDA4CB
93 changed files with 1379 additions and 1212 deletions
Download patch file Download diff file Expand all files Collapse all files

36
aiogram/webhook/aiohttp_server.py
View file

@ -2,7 +2,8 @@ import asyncio
import secrets
from abc import ABC, abstractmethod
from asyncio import Transport
from typing import Any, Awaitable, Callable, Dict, Optional, Set, Tuple, cast
from collections.abc import Awaitable, Callable
from typing import TYPE_CHECKING, Any, cast
from aiohttp import JsonPayload, MultipartWriter, Payload, web
from aiohttp.typedefs import Handler
@ -12,9 +13,11 @@ from aiohttp.web_middlewares import middleware
from aiogram import Bot, Dispatcher, loggers
from aiogram.methods import TelegramMethod
from aiogram.methods.base import TelegramType
from aiogram.types import InputFile
from aiogram.webhook.security import IPFilter
if TYPE_CHECKING:
from aiogram.types import InputFile
def setup_application(app: Application, dispatcher: Dispatcher, /, **kwargs: Any) -> None:
"""
@ -42,7 +45,7 @@ def setup_application(app: Application, dispatcher: Dispatcher, /, **kwargs: Any
app.on_shutdown.append(on_shutdown)
def check_ip(ip_filter: IPFilter, request: web.Request) -> Tuple[str, bool]:
def check_ip(ip_filter: IPFilter, request: web.Request) -> tuple[str, bool]:
# Try to resolve client IP over reverse proxy
if forwarded_for := request.headers.get("X-Forwarded-For", ""):
# Get the left-most ip when there is multiple ips
@ -98,7 +101,7 @@ class BaseRequestHandler(ABC):
self.dispatcher = dispatcher
self.handle_in_background = handle_in_background
self.data = data
self._background_feed_update_tasks: Set[asyncio.Task[Any]] = set()
self._background_feed_update_tasks: set[asyncio.Task[Any]] = set()
def register(self, app: Application, /, path: str, **kwargs: Any) -> None:
"""
@ -128,13 +131,12 @@ class BaseRequestHandler(ABC):
:param request:
:return: Bot instance
"""
pass
@abstractmethod
def verify_secret(self, telegram_secret_token: str, bot: Bot) -> bool:
pass
async def _background_feed_update(self, bot: Bot, update: Dict[str, Any]) -> None:
async def _background_feed_update(self, bot: Bot, update: dict[str, Any]) -> None:
result = await self.dispatcher.feed_raw_update(bot=bot, update=update, **self.data)
if isinstance(result, TelegramMethod):
await self.dispatcher.silent_call_request(bot=bot, result=result)
@ -142,15 +144,18 @@ class BaseRequestHandler(ABC):
async def _handle_request_background(self, bot: Bot, request: web.Request) -> web.Response:
feed_update_task = asyncio.create_task(
self._background_feed_update(
bot=bot, update=await request.json(loads=bot.session.json_loads)
)
bot=bot,
update=await request.json(loads=bot.session.json_loads),
),
)
self._background_feed_update_tasks.add(feed_update_task)
feed_update_task.add_done_callback(self._background_feed_update_tasks.discard)
return web.json_response({}, dumps=bot.session.json_dumps)
def _build_response_writer(
self, bot: Bot, result: Optional[TelegramMethod[TelegramType]]
self,
bot: Bot,
result: TelegramMethod[TelegramType] | None,
) -> Payload:
if not result:
# we need to return something "empty"
@ -166,7 +171,7 @@ class BaseRequestHandler(ABC):
payload = writer.append(result.__api_method__)
payload.set_content_disposition("form-data", name="method")
files: Dict[str, InputFile] = {}
files: dict[str, InputFile] = {}
for key, value in result.model_dump(warnings=False).items():
value = bot.session.prepare_value(value, bot=bot, files=files)
if not value:
@ -185,7 +190,7 @@ class BaseRequestHandler(ABC):
return writer
async def _handle_request(self, bot: Bot, request: web.Request) -> web.Response:
result: Optional[TelegramMethod[Any]] = await self.dispatcher.feed_webhook_update(
result: TelegramMethod[Any] | None = await self.dispatcher.feed_webhook_update(
bot,
await request.json(loads=bot.session.json_loads),
**self.data,
@ -209,7 +214,7 @@ class SimpleRequestHandler(BaseRequestHandler):
dispatcher: Dispatcher,
bot: Bot,
handle_in_background: bool = True,
secret_token: Optional[str] = None,
secret_token: str | None = None,
**data: Any,
) -> None:
"""
@ -244,7 +249,7 @@ class TokenBasedRequestHandler(BaseRequestHandler):
self,
dispatcher: Dispatcher,
handle_in_background: bool = True,
bot_settings: Optional[Dict[str, Any]] = None,
bot_settings: dict[str, Any] | None = None,
**data: Any,
) -> None:
"""
@ -265,7 +270,7 @@ class TokenBasedRequestHandler(BaseRequestHandler):
if bot_settings is None:
bot_settings = {}
self.bot_settings = bot_settings
self.bots: Dict[str, Bot] = {}
self.bots: dict[str, Bot] = {}
def verify_secret(self, telegram_secret_token: str, bot: Bot) -> bool:
return True
@ -283,7 +288,8 @@ class TokenBasedRequestHandler(BaseRequestHandler):
:param kwargs:
"""
if "{bot_token}" not in path:
raise ValueError("Path should contains '{bot_token}' substring")
msg = "Path should contains '{bot_token}' substring"
raise ValueError(msg)
super().register(app, path=path, **kwargs)
async def resolve_bot(self, request: web.Request) -> Bot:

17
aiogram/webhook/security.py
View file

@ -1,5 +1,5 @@
from collections.abc import Sequence
from ipaddress import IPv4Address, IPv4Network
from typing import Optional, Sequence, Set, Union
DEFAULT_TELEGRAM_NETWORKS = [
IPv4Network("149.154.160.0/20"),
@ -8,17 +8,17 @@ DEFAULT_TELEGRAM_NETWORKS = [
class IPFilter:
def __init__(self, ips: Optional[Sequence[Union[str, IPv4Network, IPv4Address]]] = None):
self._allowed_ips: Set[IPv4Address] = set()
def __init__(self, ips: Sequence[str | IPv4Network | IPv4Address] | None = None):
self._allowed_ips: set[IPv4Address] = set()
if ips:
self.allow(*ips)
def allow(self, *ips: Union[str, IPv4Network, IPv4Address]) -> None:
def allow(self, *ips: str | IPv4Network | IPv4Address) -> None:
for ip in ips:
self.allow_ip(ip)
def allow_ip(self, ip: Union[str, IPv4Network, IPv4Address]) -> None:
def allow_ip(self, ip: str | IPv4Network | IPv4Address) -> None:
if isinstance(ip, str):
ip = IPv4Network(ip) if "/" in ip else IPv4Address(ip)
if isinstance(ip, IPv4Address):
@ -26,16 +26,17 @@ class IPFilter:
elif isinstance(ip, IPv4Network):
self._allowed_ips.update(ip.hosts())
else:
raise ValueError(f"Invalid type of ipaddress: {type(ip)} ('{ip}')")
msg = f"Invalid type of ipaddress: {type(ip)} ('{ip}')"
raise ValueError(msg)
@classmethod
def default(cls) -> "IPFilter":
return cls(DEFAULT_TELEGRAM_NETWORKS)
def check(self, ip: Union[str, IPv4Address]) -> bool:
def check(self, ip: str | IPv4Address) -> bool:
if not isinstance(ip, IPv4Address):
ip = IPv4Address(ip)
return ip in self._allowed_ips
def __contains__(self, item: Union[str, IPv4Address]) -> bool:
def __contains__(self, item: str | IPv4Address) -> bool:
return self.check(item)