Хочу рассказать про библиотеку aiogram_tool, которая добавляет в написание ботов дополнительные возможности. Если точнее, то речь пойдёт про Dependency Injection, RateLimit и LongCallbackData.

Сначала хочу немного рассказать, зачем она вообще была написана. Во время разработки ботов многие сталкиваются с тем, что одну и ту же логику приходится реализовывать каждый раз, в каждом новом боте. Именно эту проблему и решает aiogram_tool: он объединяет полезные инструменты, которые делают разработку быстрее. Теперь подробнее про сами инструменты.

Регистрация инструментов

Первое, с чего я начну, — регистрация инструментов. Если их много, должна быть единая точка входа, и она есть: функция aiogram_tool_setup.

def aiogram_tool_setup(
    dispatcher: Dispatcher,
    tools: Iterable[BaseTool],
) -> None:
    ...

Аргументы:

dispatcher: Dispatcher — обязательный, экземпляр диспетчера aiogram;
tools: Iterable[BaseTool] — обязательный, список инструментов.

Главная функция регистрации инструментов. Для DI передайте экземпляр класса DependTool в списке tools, для RateLimit — RateLimitTool.

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

import asyncio

from aiogram import Bot, Dispatcher

from aiogram_tool.tools.depend import DependTool
from aiogram_tool.tools.limit import RateLimitTool
from aiogram_tool.tools.setup import aiogram_tool_setup

bot = Bot("YOUR_TOKEN_HERE")
dp = Dispatcher()


async def main():
    aiogram_tool_setup(dp, [DependTool(), RateLimitTool()])
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Dependency Injection

Первый инструмент, с которого я начну, — Dependency Injection. При написании я вдохновлялся Depends из FastAPI. Зацикливаться на простых примерах не хочу, их можно увидеть в документации. Я хочу показать основные особенности моей реализации.

Объявление зависимости

Зависимость можно объявить двумя способами: через значение по умолчанию или через typing.Annotated.

from typing import Annotated

from aiogram.filters import CommandStart
from aiogram.types import Message

from aiogram_tool.tools.depend import Depends


async def get_user_name(context: Message) -> str:
    return context.from_user.full_name


# Через значение по умолчанию
@dp.message(CommandStart())
async def start_handler(message: Message, name: str = Depends(get_user_name)):
    await message.answer(f"Привет, {name}!")


# Через Annotated
@dp.message(CommandStart())
async def start_handler_annotated(
    message: Message,
    name: Annotated[str, Depends(get_user_name)],
):
    await message.answer(f"Привет, {name}!")

Аргументы зависимости без значения по умолчанию берутся из middleware_data, то есть из всех данных, которые aiogram передаёт в обработчик. Например, context здесь — текущее событие.

Поддерживаются async- и sync-функции, class-functor’ы, классы (их __init__ тоже получает зависимости), async-генераторы и @asynccontextmanager. Вложенные зависимости тоже работают, а циклические цепочки обнаруживаются и приводят к DependRecursionError.

Регистрация scopes

Время жизни зависимости задаётся через Scope. Все виды scope:

Scope	Описание
`Scope.TRANSIENT`	По умолчанию. Зависимость вызывается каждый раз
`Scope.REQUEST`	Результат кэшируется на время обработки одного апдейта
`Scope.SINGLETON`	Результат кэшируется на всё время работы приложения

Задавать scope можно через ScopeRegistry:

import secrets

from aiogram_tool.tools.depend import Scope, ScopeRegistry, DependTool
from aiogram_tool.tools.setup import aiogram_tool_setup

scope_registry = ScopeRegistry()


@scope_registry(Scope.SINGLETON)
async def get_app_config() -> str:
    return "APP_CONFIG_V1"


@scope_registry(Scope.REQUEST)
async def get_request_id() -> int:
    return secrets.randbits(20)


aiogram_tool_setup(dp, [DependTool(scope_registry=scope_registry)])

Или напрямую в Depends:

@dp.message(CommandStart())
async def start_handler(
    message: Message,
    config: str = Depends(get_app_config, scope=Scope.SINGLETON),
): ...

Scope, переданный в Depends, имеет приоритет над scope, зарегистрированным в ScopeRegistry.

Самое главное — передать экземпляр ScopeRegistry в DependTool, иначе регистрации не будут учтены.

DependExit — отмена вызова обработчика

Если зависимость выбрасывает исключение DependExit, обработчик не вызывается:

from aiogram_tool.tools.depend import DependExit


async def verify_user_access(context: Message) -> None:
    if context.from_user.id != 123456789:
        await context.answer("You are not admin!")
        raise DependExit()


@dp.message(CommandStart())
async def start_handler(
    message: Message,
    _=Depends(verify_user_access),
):
    await message.answer("Welcome admin!")

DependFilter — вызов зависимости на уровне фильтра

Чтобы не создавать неиспользуемый аргумент через _, можно воспользоваться DependFilter. Он выполняет зависимость на этапе фильтров: если она выбрасывает DependExit, фильтр возвращает False, и обработчик не вызывается.

from aiogram_tool.tools.depend import DependFilter


@dp.message(
    CommandStart(),
    DependFilter(Depends(verify_user_access)),
)
async def start_handler(message: Message):
    await message.answer("Welcome admin!")

Dependency override

Поддерживается подмена зависимостей, это удобно для тестов и локальной разработки. Зависимости передаются в обработчик через значение по умолчанию или через Annotated. Подмена задаётся в DependTool:

async def get_external_data():
    return "REAL_API_DATA"


async def get_mocked_data():
    return "MOCKED_DATA"


depend_tool = DependTool(
    dependency_override={
        get_external_data: Depends(get_mocked_data),
    }
)

Подробнее о всех возможностях можно узнать в документации.

RateLimit — ограничение частоты запросов

Задача простая: ограничить, сколько раз пользователь может вызвать обработчик за единицу времени.

Инструмент построен на двух классах:

RateLimitTool регистрируется один раз через aiogram_tool_setup и хранит настройки по умолчанию для всех обработчиков: storage (где хранятся лимиты) и answer_callback (что ответить при превышении);
RateLimitFilter добавляется в конкретный обработчик. В него тоже можно передать storage и answer_callback, и тогда они будут использоваться только этим обработчиком.

При вызове фильтр ищет RateLimitTool в dispatcher.workflow_data, строит уникальный ключ, и под блокировкой хранилища проверяет лимит. Блокировка делает проверку атомарной даже при конкурентных апдейтах. Если лимит не исчерпан, обработчик вызывается. Если исчерпан, вызывается answer_callback, а обработчик не вызывается.

Быстрый старт:

from datetime import timedelta

from aiogram.filters import Command
from aiogram.types import Message

from aiogram_tool.tools.limit import RateLimitFilter, RateLimitTool
from aiogram_tool.tools.limit.rate_limit import FixedWindowRateLimit
from aiogram_tool.tools.setup import aiogram_tool_setup


@dp.message(
    # Лимит: 3 запроса за 10 секунд на пользователя
    Command("ping"),
    RateLimitFilter(
        rate_limit=FixedWindowRateLimit(requests=3, time=timedelta(seconds=10))
    ),
)
async def ping_handler(message: Message):
    await message.answer("Pong!")


aiogram_tool_setup(dp, [RateLimitTool()])

Алгоритмы

В инструменте есть три алгоритма ограничения.

FixedWindowRateLimit — фиксированное окно. Окно открывается с первого запроса и закрывается через time. Внутри окна разрешено не более requests запросов, после чего все запросы отклоняются до конца окна.

RateLimitFilter(rate_limit=FixedWindowRateLimit(requests=5, time=timedelta(seconds=60)))

У этого алгоритма есть особенность: возможны всплески на границе окна. Пользователь может отправить requests запросов в конце одного окна и столько же в начале следующего.

SlidingWindowRateLimit — скользящее окно. Хранятся временные метки последних запросов, и лимит — не более requests запросов за любые time. Всплесков на границе окна нет, но и данных в хранилище больше.

RateLimitFilter(
    rate_limit=SlidingWindowRateLimit(requests=5, time=timedelta(seconds=60))
)

TokenBucketRateLimit — токеновое ведро. В ведре накапливаются токены (не больше bucket_size), каждый запрос расходует один токен, а пополнение идёт со скоростью refill_tokens каждые refill_time. Алгоритм разрешает кратковременные всплески за счёт накопленных токенов и при этом ограничивает среднюю частоту.

RateLimitFilter(
    rate_limit=TokenBucketRateLimit(
        bucket_size=5,  # максимальное число токенов
        current_tokens=5,  # начальное число токенов
        refill_time=timedelta(seconds=5),  # интервал пополнения
        refill_tokens=1,  # +1 токен каждые 5 секунд
    )
)

Алгоритм	Аргументы	Особенности
`FixedWindowRateLimit`	`requests`, `time`	Простой, минимум данных; возможны всплески на границе окна
`SlidingWindowRateLimit`	`requests`, `time`	Точный, без всплесков; хранит список временных меток
`TokenBucketRateLimit`	`bucket_size`, `current_tokens`, `refill_time`, `refill_tokens`	Всплески разрешены, средняя частота ограничена скоростью пополнения

Персональные и глобальные лимиты

По умолчанию лимит действует на каждого пользователя отдельно: в ключ входит user_id. Если указать all_users=True, лимит станет общим для всех пользователей:

@dp.message(
    Command("start"),
    RateLimitFilter(
        rate_limit=SlidingWindowRateLimit(requests=10, time=timedelta(minutes=1)),
        all_users=True,  # лимит общий для всех
        key="global_start",  # свой ключ вместо имени обработчика
    ),
)
async def start_handler(message: Message):
    await message.answer("10 запросов в минуту на всех")

Свой ответ при превышении лимита

По умолчанию пользователь получает сообщение вида Next request after 7.0 seconds.. Чтобы изменить его, достаточно унаследоваться от RateLimitAnswer и переопределить __call__:

from datetime import timedelta

from aiogram.types import TelegramObject

from aiogram_tool.tools.limit import RateLimitAnswer, RateLimitTool


class CustomLimitAnswer(RateLimitAnswer):
    async def __call__(
        self, event: TelegramObject, window_time: timedelta, retry_after: timedelta
    ) -> None:
        await event.answer(
            text=f"🚫 Слишком часто! Повторите через {retry_after.total_seconds():.1f} сек."
        )


rate_limit_tool = RateLimitTool(answer_callback=CustomLimitAnswer())

Хранилища

Лимиты хранятся в хранилище с поддержкой блокировок:

Хранилище	Описание
`MemoryLockStorage`	По умолчанию. Данные в памяти процесса, сбрасываются при перезапуске
`FileLockStorage`	Данные в файле, переживают перезапуск
`AsyncRedisLockStorage`	Данные в Redis, подходят для нескольких инстансов бота

from redis.asyncio import Redis as AsyncRedis

from aiogram_tool.storage import AsyncRedisLockStorage
from aiogram_tool.tools.limit import RateLimitTool

rate_limit_tool = RateLimitTool(storage=AsyncRedisLockStorage(redis=AsyncRedis()))

Хранилище можно переопределить и для отдельного обработчика, передав storage в RateLimitFilter.

Подводные камни

Есть три момента, на которые стоит обратить внимание.

События без атрибута from_user не поддерживаются, будет выброшено TypeError.
По умолчанию ключ формируется из имени модуля и функции обработчика. Если имена двух обработчиков совпадут, они будут делить один лимит, поэтому для надёжности задавайте key явно.
Фильтр получает RateLimitTool через диспетчер, который ищется в данных апдейта. При обычном polling это работает автоматически. Если апдейты обрабатываются вручную (например, через webhook и feed_update), передайте в фильтр dispatcher=..., иначе будет выброшено ValueError("Dispatcher not found").

Все примеры и подробности есть в документации.

LongCallbackData — длинный callback_data у InlineKeyboardButton

Telegram ограничивает размер атрибута callback_data инлайн-кнопки 64 байтами. Если сериализованные данные превышают этот лимит, aiogram выбрасывает ошибку ValueError: Resulted callback data is too long!.

LongCallbackData решает эту проблему: слишком «длинные» данные автоматически сохраняются в хранилище, а в кнопку упаковывается короткий уникальный идентификатор. Когда пользователь нажимает на кнопку, данные прозрачно восстанавливаются из хранилища. API остаётся таким же, как у стандартного CallbackData в aiogram.

from aiogram import F
from aiogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup
from aiogram.filters import CommandStart

from aiogram_tool.tools.callback_data import LongCallbackData


class MyLongData(LongCallbackData, prefix="mydata"):
    mode: str
    payload: str


@dp.message(CommandStart())
async def start_handler(message: Message):
    # Короткие данные упаковываются как обычно
    short_cb = await MyLongData(mode="short", payload="Hello!").pack_long()
    # Длинные данные (больше 64 байт) автоматически сохраняются в хранилище
    long_cb = await MyLongData(mode="long", payload="A" * 200).pack_long()

    await message.answer(
        "Выберите действие:",
        reply_markup=InlineKeyboardMarkup(
            inline_keyboard=[
                [InlineKeyboardButton(text="Короткие данные", callback_data=short_cb)],
                [InlineKeyboardButton(text="Длинные данные", callback_data=long_cb)],
            ]
        ),
    )


# Фильтры работают так же, как в стандартном aiogram
@dp.callback_query(MyLongData.filter(F.mode == "long"))
async def process_long_data(query: CallbackQuery, callback_data: MyLongData):
    await query.answer(text=f"Длина payload: {len(callback_data.payload)}")

Также можно контролировать хранилище, в которое сохраняются длинные данные. Для этого переопределите атрибут класса _storage, например на MemoryStorage, FileStorage или AsyncRedisStorage:

from redis.asyncio import Redis as AsyncRedis

from aiogram_tool.storage import AsyncRedisStorage

redis_storage = AsyncRedisStorage(
    redis=AsyncRedis(host="localhost", port=6379, decode_responses=True),
    expire=3600,  # данные живут 1 час
)


class PersistentData(LongCallbackData, prefix="redis"):
    _storage = redis_storage

    user_id: int
    big_context: str

Данные при этом переживают перезапуск бота.

Если callback_data не найдена в хранилище (например, бот перезапустился), вызывается _answer_callback, то есть __call__ класса CallbackDataAnswer. По умолчанию он показывает alert «Button expired». Чтобы задать свой ответ, наследуйтесь от CallbackDataAnswer и переопределите __call__:

from aiogram.types import CallbackQuery

from aiogram_tool.tools.callback_data import CallbackDataAnswer


class MyExpiredAnswer(CallbackDataAnswer):
    async def __call__(self, query: CallbackQuery) -> None:
        await query.message.answer("Кнопка устарела, отправьте /start заново.")
        await query.answer()


class MyLongData(LongCallbackData, prefix="mydata"):
    _answer_callback = MyExpiredAnswer()

    mode: str
    payload: str

Для LongCallbackData aiogram_tool_setup не нужен, класс работает сам по себе.

Итоги

Все три инструмента решают типичные задачи, с которыми сталкивается каждый, кто пишет ботов на aiogram: регистрация зависимостей, ограничение частоты запросов и длинные данные в кнопках. Исходный код, документация и примеры доступны в репозитории. Буду рад вашим замечаниям и идеям в issues.

Инструменты для aiogram