[SOURCE] Throttled декоратор/миддлварь для Aiogram 3.

Предистория

Решил я значит пересесть с Aiogram 2 на Aiogram 3 с лёту, ибо каждый раз искать документацию для Aiogram 2.25 я конкретно з**бался, ну значит пишу се бота, и появилась необходимость в троттл декораторе (для тех, кто не в курсе, это примитивный анти-спам: https://docs.aiogram.dev/en/v2.25.1/dispatcher/index.html#aiogram.Dispatcher.throttled), и тут я вижу что в снипеттах его нет, полез в документацию уже нового Aiogram 3 и вижу это:

Я немного прихуел, но не испугался и пошёл писать его с нуля . И получилось, на мой взгляд, настолько хорошо что решил поделится с вами.

К делу

Тут я опишу полный тутор как и что работает, а для тех, для кого сэкономленное время играет важную роль - листайте вниз.

Шедевроимпорты:

from typing import Any, Callable, Dict, Awaitable # Необязательно, если пишете для себя и один раз
from aiogram import BaseMiddleware
from aiogram.types import TelegramObject
from cachetools import TTLCache

Python

from typing import Any, Callable, Dict, Awaitable # Необязательно, если пишете для себя и один раз

from aiogram import BaseMiddleware

from aiogram.types import TelegramObject

from cachetools import TTLCache

# Объявляем функцию будущего декоратора,
# Роль которого по-сути тупо передать аргументы в функцию.
# Тут даже объяснять нечего, для тех кто изучал декораторы поймёт что тут написано
def throttled(rate: int, on_throttle: Callable | None = None):
def decorator(func):
setattr(func, "rate", rate)
setattr(func, "on_throttle", on_throttle)
return func

return decorator

Python

# Объявляем функцию будущего декоратора,

# Роль которого по-сути тупо передать аргументы в функцию.

# Тут даже объяснять нечего, для тех кто изучал декораторы поймёт что тут написано

def throttled(rate: int, on_throttle: Callable | None = None):

    def decorator(func):

        setattr(func, "rate", rate)

        setattr(func, "on_throttle", on_throttle)

        return func

 

    return decorator

Самый жир

class ThrottlingMiddleware(BaseMiddleware):
def __init__(self):
# self.cache = TTLCache(maxsize=10_000, ttl=self.delay)
self.caches = dict()

async def __call__(
self,
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
event: TelegramObject,
data: Dict[str, Any],
) -> Any:

# Получаем функцию, которая находится под декоратором/роутером, т.е.
# @router.message()
# def Echo(msg: Message) <- Адресс в памяти (объект) этой функции мы и получаем
decorated_func = data["handler"].callback

# Пытаемся получить переданные нашим новым декоратором (@throttled) аргументы
# Соотв. если они есть - получим значение, нет - None. Детальнее читайте про getattr() метод.
rate = getattr(decorated_func, "rate", None)
on_throttle = getattr(decorated_func, "on_throttle", None)

# Проверяем что:
# 1 - rate не None/0
# 2 - rate является объектом <int>
if rate and isinstance(rate, int):

# Тут стоит объяснить как работает TTLCache
# Это по-сути обычный словарь, который сам удаляет элементы,
# если они там находятся n и более секунд
# И если TTLCache декорируемой функции ещё не в словаре - добавляем
if id(decorated_func) not in self.caches:
self.caches[id(decorated_func)] = TTLCache(maxsize=10_000, ttl=rate)

# Далее, проверяем, находится ли наш юзер в TTLCache функции
# Если да, значит он достиг рейт-лимита и должен получить пизды.
# И да, забыл сказать, этот миддлварь не будет работать в общих чатах
# Мне лень было делать проверку на тип чата, так что довольствуйтесь чем есть
# У кого руки не из жопы - переделают
if event.from_user.id in self.caches[id(decorated_func)].keys():
# Проверяем что переданное значение в on_throttle
# является функцией (не только, но не суть)
# И если это так, то вызываем её, вместо функции хендлера/роутера
# Ну а если нет то просто обрываем дальнейший вызов функции хендлера/роутера
if callable(on_throttle):
return await on_throttle(event, data)
else:
return
else:
# Ну а если пользователь не в TTLCache, то он послушный мальчик
# и бить его не надо, только запихнуть в TTL по-новой, что-бы следить дальше.
self.caches[id(decorated_func)][event.chat.id] = event.chat.id

# Ну и типикал вызов дальнейшей обработки хендлера
return await handler(event, data)
else:
# rate - не задан, анти-спам на эту функцию выключен
# значит просто вызываем дальнейший ход хендлера.
return await handler(event, data)

Python

class ThrottlingMiddleware(BaseMiddleware):

    def __init__(self):

        # self.cache = TTLCache(maxsize=10_000, ttl=self.delay)

        self.caches = dict()



    async def __call__(

            self,

            handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],

            event: TelegramObject,

            data: Dict[str, Any],

    ) -> Any:

 

        # Получаем функцию, которая находится под декоратором/роутером, т.е.

        # @router.message()

        # def Echo(msg: Message) <- Адресс в памяти (объект) этой функции мы и получаем

        decorated_func = data["handler"].callback



        # Пытаемся получить переданные нашим новым декоратором (@throttled) аргументы

        # Соотв. если они есть - получим значение, нет - None. Детальнее читайте про getattr() метод.

        rate = getattr(decorated_func, "rate", None)

        on_throttle = getattr(decorated_func, "on_throttle", None)

 

        # Проверяем что:

        # 1 - rate не None/0

        # 2 - rate является объектом <int>

        if rate and isinstance(rate, int):

     

            # Тут стоит объяснить как работает TTLCache

            # Это по-сути обычный словарь, который сам удаляет элементы,

            # если они там находятся n и более секунд

            # И если TTLCache декорируемой функции ещё не в словаре - добавляем

            if id(decorated_func) not in self.caches:

                self.caches[id(decorated_func)] = TTLCache(maxsize=10_000, ttl=rate)

     

            # Далее, проверяем, находится ли наш юзер в TTLCache функции

            # Если да, значит он достиг рейт-лимита и должен получить пизды.

            # И да, забыл сказать, этот миддлварь не будет работать в общих чатах

            # Мне лень было делать проверку на тип чата, так что довольствуйтесь чем есть

            # У кого руки не из жопы - переделают

            if event.from_user.id in self.caches[id(decorated_func)].keys():

                # Проверяем что переданное значение в on_throttle

                # является функцией (не только, но не суть)

                # И если это так, то вызываем её, вместо функции хендлера/роутера

                # Ну а если нет то просто обрываем дальнейший вызов функции хендлера/роутера

                if callable(on_throttle):

                    return await on_throttle(event, data)

                else:

                    return

            else:

                # Ну а если пользователь не в TTLCache, то он послушный мальчик

                # и бить его не надо, только запихнуть в TTL по-новой, что-бы следить дальше.

                self.caches[id(decorated_func)][event.chat.id] = event.chat.id

         

                # Ну и типикал вызов дальнейшей обработки хендлера

                return await handler(event, data)

        else:

            # rate - не задан, анти-спам на эту функцию выключен

            # значит просто вызываем дальнейший ход хендлера.

            return await handler(event, data)

Весь код

from typing import Any, Callable, Dict, Awaitable
from aiogram import BaseMiddleware
from aiogram.types import TelegramObject
from cachetools import TTLCache

def throttled(rate: int, on_throttle: Callable | None = None):
"""Throttled decorator, must be above router/dispatcher decorator!!!

Args:
rate (int): Determines how long the user will have to wait (seconds) to use again.
on_throttle (Callable or None, optional): Callback function if rate limit exceed. Defaults to None.
"""
def decorator(func):
setattr(func, "rate", rate)
setattr(func, "on_throttle", on_throttle)
return func

return decorator

class ThrottlingMiddleware(BaseMiddleware):
"""
Throttling middleware which is used as an anti-spam tool
1. Initialize and attach to router/dispatcher
2. Use @throttled(rate, on_throttle) above router decorator
"""
def __init__(self):
# self.cache = TTLCache(maxsize=10_000, ttl=self.delay)
self.caches = dict()

async def __call__(
self,
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
event: TelegramObject,
data: Dict[str, Any],
) -> Any:

decorated_func = data["handler"].callback
rate = getattr(decorated_func, "rate", None)
on_throttle = getattr(decorated_func, "on_throttle", None)

if rate and isinstance(rate, int) and rate > 0: # Check if rate arg passed and passed correctly (decorator check)
if id(decorated_func) not in self.caches: # Check if func TTL already in dict. If not - create it.
self.caches[id(decorated_func)] = TTLCache(maxsize=10_000, ttl=rate)

if event.from_user.id in self.caches[id(decorated_func)].keys():
if callable(on_throttle):
return await on_throttle(event, data)
else:
return
else:
self.caches[id(decorated_func)][event.chat.id] = event.chat.id
return await handler(event, data)
else:
return await handler(event, data)[FONT=inherit][COLOR=inherit]

Python

from typing import Any, Callable, Dict, Awaitable

from aiogram import BaseMiddleware

from aiogram.types import TelegramObject

from cachetools import TTLCache



def throttled(rate: int, on_throttle: Callable | None = None):

    """Throttled decorator, must be above router/dispatcher decorator!!!



    Args:

        rate (int): Determines how long the user will have to wait (seconds) to use again.

        on_throttle (Callable or None, optional): Callback function if rate limit exceed. Defaults to None.

    """

    def decorator(func):

        setattr(func, "rate", rate)

        setattr(func, "on_throttle", on_throttle)

        return func

 

    return decorator



class ThrottlingMiddleware(BaseMiddleware):

    """

    Throttling middleware which is used as an anti-spam tool

    1. Initialize and attach to router/dispatcher

    2. Use @throttled(rate, on_throttle) above router decorator

    """

    def __init__(self):

        # self.cache = TTLCache(maxsize=10_000, ttl=self.delay)

        self.caches = dict()



    async def __call__(

            self,

            handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],

            event: TelegramObject,

            data: Dict[str, Any],

    ) -> Any:

 

        decorated_func = data["handler"].callback

        rate = getattr(decorated_func, "rate", None)

        on_throttle = getattr(decorated_func, "on_throttle", None)



        if rate and isinstance(rate, int) and rate > 0: # Check if rate arg passed and passed correctly (decorator check)

            if id(decorated_func) not in self.caches: # Check if func TTL already in dict. If not - create it.

                self.caches[id(decorated_func)] = TTLCache(maxsize=10_000, ttl=rate)



            if event.from_user.id in self.caches[id(decorated_func)].keys():

                if callable(on_throttle):

                    return await on_throttle(event, data)

                else:

                    return

            else:

                self.caches[id(decorated_func)][event.chat.id] = event.chat.id

                return await handler(event, data)

        else:

            return await handler(event, data)[FONT=inherit][COLOR=inherit]

[/COLOR][/FONT]

Надеюсь хоть кому-то это будет да полезно, если да, жду +реп ну или копейкой отблагодарите
Да, не идеально и не самое чистое решение, но для личного пользования пойдёт
Так-же забыл добавить, что наш новый декоратор должен быть строго выше декоратора-хендлера
​

UPD:
В соотв. с новыми положениями и принципами Python (если не ошибаюсь 3.13.x) нужно чуть-чуть изменить аннотации и вместо "or" использовать бинарный оператор "или"

Код:
def throttled(rate: int, on_throttle: Callable or None = None):
def throttled(rate: int, on_throttle: Callable | None = None):

UPD 2:
Чуть не забыл, чтобы это работало - нужно закрепить наш миддлварь за роутером и нужным событием, я предпочитаю это делать в месте где я объединяю все роутеры в один:

Код:

CODE_1

....
from middlewares.throttled import ThrottlingMiddleware

router = Router()
router.include_router(mr)
router.include_router(pr)
router.include_router(ar)

router.message.middleware(ThrottlingMiddleware()) # На сообщения
router.callback_query.middleware(ThrottlingMiddleware()) # На кнопки

__all__ = ["router"]

Python

....

from middlewares.throttled import ThrottlingMiddleware



router = Router()

router.include_router(mr)

router.include_router(pr)

router.include_router(ar)



router.message.middleware(ThrottlingMiddleware()) # На сообщения

router.callback_query.middleware(ThrottlingMiddleware()) # На кнопки



__all__ = ["router"]

И как верно подсказал reTelegram лучше использовать не

⁡event.chat.id

⁡, а вместо этого:

⁡event.from_user.id

⁡, что позволит отслеживать не сообщения от "чата" а от определённого пользователя и что позволит в последствии использовать этот миддлварь даже в группах/многопользовательских чатах.