Middleware в aiogram: антиспам, логирование и общие данные

Ваш бот из урока про обработчики оброс командами, из урока про асинхронность он научился не замирать — и вот в него пришли люди. Все сразу. Пятнадцать сообщений в секунду из одного чата, рассылка спама, попытки уронить /calc строкой «+++». Хендлеры тут не помогут: они про «что отвечать», а вам нужен код про «пропускать ли вообще». Такой код называется middleware — и это последний архитектурный инструмент aiogram, которого у вас ещё нет.

В этом уроке разберём механику на чистом Python — она запускается прямо на странице, — а затем покажем настоящий код на aiogram 3 с классом BaseMiddleware. Соберём три классических middleware: троттлинг-антиспам, логирование апдейтов и передачу общих данных в хендлеры.

Что такое middleware и чем он отличается от хендлера?

Middleware — это код, который стоит между апдейтом и хендлером: он видит каждое сообщение раньше вашего обработчика. Диспетчер aiogram получает апдейт от Telegram и ведёт его по цепочке: сначала middleware, потом фильтры, потом хендлер. Middleware может пропустить апдейт дальше, изменить его, пополнить словарь данных — или остановить цепочку, и тогда хендлер даже не узнает, что что-то приходило.

Правило большое: хендлеры — про бизнес-логику, middleware — про политику. «Считать сумму двух чисел» — хендлер. «Не больше одного сообщения в полсекунды от одного человека» — middleware. Если смешать их, антиспам-проверки расползутся по каждому хендлеру, и добавить новую проверку будет значит переписать половину бота.

• Троттлинг — не давать одному пользователю завалить бота (сегодняшняя тема);
• Логирование — записывать каждый апдейт до того, как его увидят хендлеры;
• Общие данные — положить в data подключение к базе, конфиг, сессию — один раз на весь бот;
• Аутентификация — пускать только пользователей из белого списка, остальным сразу отказ;
• Язык и локаль — определить язык пользователя один раз, а не в каждом хендлере.

Как устроен лук middleware на чистом Python?

Внутри aiogram нет никакой магии — есть функции-обёртки. Обёртка получает хендлер и возвращает новый хендлер, который делает что-то до и после вызова оригинала. Так рождается «лук»: каждый слой middleware заворачивает всё, что внутри него. Запустите и посмотрите:

def with_logging(handler):
    def wrapper(event):
        print("-> апдейт:", event["text"])
        result = handler(event)
        print("<- ответ отправлен")
        return result
    return wrapper

def handle_message(event):
    return "Эхо: " + event["text"]

handler = with_logging(handle_message)
print(handler({"user_id": 1, "text": "привет"}))

-> апдейт: привет
<- ответ отправлен
Эхо: привет

Разберём по строчкам. with_logging(handler) — фабрика обёрток: она не выполняет ничего сама, а возвращает новую функцию wrapper. Именно wrapper становится обработчиком: сначала печатает строку до, потом вызывает настоящий handler(event), потом печатает строку после — и не забывает вернуть результат. Пока это просто логирование, но на этом каркасе строится весь антиспам: между «до» и «после» можно поставить любую проверку.

А теперь наденем на хендлер два слоя сразу и посмотрим на порядок срабатывания:

def logging_mw(handler):
    def wrapper(event, data):
        print("[лог] апдейт от", event["user_id"])
        return handler(event, data)
    return wrapper

def throttling_mw(handler):
    def wrapper(event, data):
        print("[антиспам] проверяю частоту")
        return handler(event, data)
    return wrapper

def handle(event, data):
    print("[хендлер] отвечаю на:", event["text"])
    return "Готово"

pipeline = logging_mw(throttling_mw(handle))
print(pipeline({"user_id": 42, "text": "/start"}, {}))

[лог] апдейт от 42
[антиспам] проверяю частоту
[хендлер] отвечаю на: /start
Готово

Смотрите, что произошло: logging_mw(throttling_mw(handle)) — первый зарегистрированный middleware оказался снаружи. Выполнение идёт сверху вниз: лог, потом антиспам, потом хендлер. Это и есть конвейер aiogram — регистрация в коде бота означает ровно вот это заворачивание в лук. Порядок важен: лог хочется видеть даже для отброшенного спама, поэтому лог-слой ставят раньше троттлинга.

Как защитить бота от флуда: троттлинг-антиспам?

Флуд — это когда один пользователь шлёт сообщения быстрее, чем бот успевает их осмысленно обрабатывать. Классическое решение — троттлинг: ограничение частоты. Для каждого пользователя храним время последнего пропущенного сообщения; новое сообщение проходит, только если прошло больше лимита. Отдельный нюанс: у самого Telegram есть собственные лимиты, и ваш троттлинг — про то, чтобы до них дело не доходило. Живому человеку задержка в полсекунды незаметна, а ботоферме вы говорите «нет» ещё на входе. Вот троттлинг на чистом Python — время приходит внутри апдейта, значения фиксированные, чтобы вывод был воспроизводим:

# Словарь user_id -> время последнего сообщения (как у настоящего троттлинга)
last_seen = {}
LIMIT = 1.0  # минимум одна секунда между сообщениями

def throttle(handler):
    def wrapper(event):
        user = event["user_id"]
        prev = last_seen.get(user)
        if prev is not None and event["time"] - prev < LIMIT:
            print("ФЛУД: пропущено от", user)
            return None  # хендлер даже не вызывается
        last_seen[user] = event["time"]
        return handler(event)
    return wrapper

def handle(event):
    print("Хендлер обработал сообщение от", event["user_id"])

pipeline = throttle(handle)

# время приходит вместе с апдейтом; значения фиксированные, из теста
pipeline({"user_id": 1, "time": 100.5})
pipeline({"user_id": 1, "time": 100.8})
pipeline({"user_id": 1, "time": 102.0})
pipeline({"user_id": 2, "time": 100.6})

Хендлер обработал сообщение от 1
ФЛУД: пропущено от 1
Хендлер обработал сообщение от 1
Хендлер обработал сообщение от 2

Три вещи, на которые стоит посмотреть. Первая: last_seen.get(user) вместо last_seen[user] — нового пользователя в словаре ещё нет, и get честно вернёт None вместо падения с KeyError. Вторая: флуд-сообщение не роняет бота и не отвечает ошибкой — оно просто не доходит до хендлера (return None). Третья: пользователь номер 2 со своим временем проходит свободно — словарь разделяет пользователей, у каждого свой лимит.

Теперь тот же троттлинг в настоящем aiogram. Класс наследуется от BaseMiddleware и реализует асинхронный метод __call__ с той же самой сигнатурой, которую вы только что видели у wrapper: хендлер, апдейт, словарь data.

import time
from collections.abc import Awaitable, Callable
from typing import Any

from aiogram import BaseMiddleware
from aiogram.types import TelegramObject


class ThrottlingMiddleware(BaseMiddleware):
    def __init__(self, limit: float = 0.5):
        self.limit = limit
        self.last_seen: dict[int, float] = {}

    async def __call__(
        self,
        handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
        event: TelegramObject,
        data: dict[str, Any],
    ) -> Any:
        user = event.from_user
        if user is None:  # апдейты из каналов без автора
            return await handler(event, data)
        now = time.monotonic()
        prev = self.last_seen.get(user.id)
        if prev is not None and now - prev < self.limit:
            return  # тихо пропускаем флуд, хендлер не вызывается
        self.last_seen[user.id] = now
        return await handler(event, data)


dp.message.middleware(ThrottlingMiddleware(limit=0.5))

Регистрация dp.message.middleware(...) вешает класс на обычные сообщения. У aiogram есть и второй способ — outer_middleware, который срабатывает ещё раньше, до фильтров:

from aiogram import Dispatcher

dp = Dispatcher()

# outer: до фильтров, на каждый апдейт - сюда ставят логи и общие данные
dp.update.outer_middleware(LogMiddleware())
# inner: после фильтров, только когда хендлер нашёлся - сюда ставят троттлинг
dp.message.middleware(ThrottlingMiddleware(limit=0.5))

Почему троттлинг — inner, а логи — outer? Логи нужны на каждый апдейт, даже на тот, что не совпал ни с одним фильтром. А троттлинг незачем гонять лишний раз: если всё равно нет хендлера, который отвечал бы на такой апдейт, — спамить нечего. Тонкость на будущее: time.monotonic() выбрано не случайно — эта шкала не прыгает назад при переводе часов, в отличие от time.time().

Как передавать общие данные в хендлеры через словарь data?

Вы уже заметили второй аргумент обёрток — словарь data. Это чемодан, который middleware набивают полезным грузом, а хендлер потом открывает. Хендлеру нужна база данных? Конфиг с токенами? Имя пользователя? Middleware кладёт это в data один раз — и каждый хендлер получает готовое, не зная, откуда оно взялось. Почему бы не завести глобальную переменную с базой? На живом боте это стреляет при тестах: глобал один, а окружений много — тестовое, боевое, локальное. Словарь data собирается заново на каждый апдейт, и хендлер получает ровно то, что нужно в этом окружении:

USERS = {7: "Анна", 42: "Борис"}

def inject_user(handler):
    def wrapper(event, data):
        # middleware обогащает data: хендлеру не нужно знать про USERS
        data["username"] = USERS.get(event["user_id"], "гость")
        data["is_new"] = event["user_id"] not in USERS
        return handler(event, data)
    return wrapper

def handle(event, data):
    if data["is_new"]:
        print(f"Рады знакомству, {data['username']}!")
    else:
        print(f"С возвращением, {data['username']}!")
    return "ok"

pipeline = inject_user(handle)
pipeline({"user_id": 7}, {})
pipeline({"user_id": 99}, {})

С возвращением, Анна!
Рады знакомству, гость!

Хендлер handle не знает ни про словарь USERS, ни про то, как определяется новый пользователь. Он просто читает data["username"] — а подготовку делает middleware. В настоящем aiogram это превращается в элегантную вещь: всё, что middleware положил в data, диспетчер подставляет в хендлер как именованный аргумент.

class DataContextMiddleware(BaseMiddleware):
    def __init__(self, db_path: str):
        self.db_path = db_path

    async def __call__(self, handler, event, data):
        data["db_path"] = self.db_path  # кладём в чемодан...
        return await handler(event, data)

dp.update.outer_middleware(DataContextMiddleware("expenses.db"))


@router.message(Command("report"))
async def cmd_report(message: Message, db_path: str):
    # ...и aiogram сам подставит db_path как аргумент хендлера
    ...

Как залогировать все апдейты одним middleware?

Логирование — самый простой и самый полезный middleware. Пять строк, а экономят часы: видно, какие апдейты приходят, от кого и что бот с ними сделал. Ошибки при этом ловить должен не лог-слой — для этого в следующем уроке есть отдельные инструменты:

import logging

from aiogram import BaseMiddleware

log = logging.getLogger("bot.updates")


class LogMiddleware(BaseMiddleware):
    async def __call__(self, handler, event, data):
        user = event.from_user
        who = user.id if user else "канал"
        log.info("апдейт от %s, текст: %r", who, event.text)
        return await handler(event, data)

Что дальше?

Подведём итог. Middleware — обёртка над хендлером, которая срабатывает на каждый апдейт; лук из обёрток вы собрали руками. Троттлинг хранит время последнего сообщения в словаре и молча глушит флуд. Словарь data — чемодан, из которого хендлеры достают общие данные, не зная, откуда они взялись. Лог-слой пишет каждый апдейт до того, как его увидит любой хендлер.

Остался последний инструмент надёжности: обработка ошибок. В уроке 9 разберём try/except в хендлерах, лимиты Telegram с исключением RetryAfter, модуль logging вместо print и перезапуск бота, который переживает падение. А затем всё соберётся в финальный проект — бота-трекер расходов. Если хочется освежить, как хендлеры и фильтры различают команды, — вернитесь ко второму уроку.

Хендлеры отвечают за то, что бот говорит. Middleware — за то, кого бот слушает.