Мы все используем aiogram, это по праву лучшая API обертка из существующих.

Однако, через длительное время работы с ней, я заметил, что таскаю из проекта в проект один и тот же код. Код, который делает aiogram лучше.

И потому я решил вынести весь этот код в отдельную библиотеку-плагин поверх aiogram — Raito.

Я не буду тратить Ваше время и сразу перечислю весь тот функционал, которого не хватало мне в aiogram'е (и это правильно, потому что aiogram должен отвечать только за оборачивание API, а не быть многоруким. Разделение ответственности да-да.):

• Hot-reload. Наверное, самый важный DX функционал в любом фреймворке. До текущего дня, уверен, вы перезапускали аж всего бота, чтобы увидеть минорные изменения кода.

• Кастомные роутеры. Тут приоритеты и динамичная загрузка. Вы сможете перезагружать роутеры через команду в чате, например, .rt reload ban

• Автоматическая регистрация команд в слэш меню с описаниями.

• Система ролей. Чуть ли не в каждом боте она есть, и каждый раз Вам приходилось разрабатывать с нуля. Я вынес это в библиотеку и сделал систему очень гибкой. Сможете переписать логику ролей и создать кастомные.

• Билдеры inline/reply клавиатур в двух режимах: статичные и динамичные.

• Serverless пагинация в 5 типах: inline, text, photo, list, rich.

• FSM сцены с валидацией данных (draft), и навигацией по этапам next/finish/goto/...

• wait_for() – еще один вариант FSM, для тех, кто не любит заморачиваться.

• Многие знают проблему с альбомами в Telegram. В плагине она уже решена.

• Классический throttling с выбором по режиму: bot, chat, user.

• lifespan в стиле fastapi для управления запуском и остановкой бота.

• И по мелочи: красивый логгинг, rt.retry, SuppressNotModifiedError

Фич много. Сейчас расскажу каждую по отдельности.

Но если Вам этого уже достаточно, и Вы хотите изучить всё самостоятельно:

• Документация: https://raito.readthedocs.io/ru/latest/index.html

• Исходный код: https://github.com/Aidenable/Raito

И я буду очень признателен, если Вы поставите звездочку ⭐ на репозиторий. Это мотивирует заниматься проектом и дальше!

Основы

Raito проектировался как инструмент поверх aiogram3, который ни в коем случае не вмешивается в работу самой библиотеки.

Подключить его легко:

# __main__.py
import asyncio

from aiogram import Bot, Dispatcher
from raito import Raito


async def main() -> None:
    bot = Bot(token="TOKEN")
    dispatcher = Dispatcher()
    raito = Raito(dispatcher, "src/handlers")

    await raito.setup()
    await dispatcher.start_polling(bot)


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

Стандартная конструкция запуска бота, за исключением 2 строк кода:

1. Raito(dispatcher, "src/handlers")
   В класс передается dispatcher для DI и регистрации роутеров
   "src/handlers" это путь к папке, где лежат все ваши роутеры/хэндлеры/команды. Да, Raito не предполагает разработку бота в single file. Вам нужно будет выносить все @router.on_message (и другие ивенты) в отдельные файлы.

2. raito.setup()
   Здесь производятся регистрации middleware, загрузка встроенных хэндлеров, загрузка ваших роутеров и включение watchdog

Кстати, на счет watchdog, чтобы включить hot-reload, нужно указать Raito(production=False)

Также вы можете указать свой Telegram ID в Raito(developers=[12345678]), чтобы автоматически выдать себе роль DEVELOPER и разблокировать доступ к .rt командам (чтобы узнать подробнее, напишите .rt help в чате с ботом)

Ну и напоследок, если введете Raito(enable_dangerous_commands=True), разблокируете eval/bash команды. Попробуйте написать в чате .rt load raito.system.eval, а затем .rt eval await _msg.answer("Hello")

Жизненный цикл

Я вдохновился lifespan декоратором из FastAPI для работы с ивентами on_startup и on_shutdown. Всё, что находится ДО yield выполнится при запуске бота, а всё, что ПОСЛЕ yield выполнится при его отключении.

# src/handlers/events/lifespan.py
from aiogram import Bot

from raito import Router, rt
from raito.core.raito import Raito

router = Router(name="lifespan")


@router.lifespan()
async def lifespan(bot: Bot, raito: Raito):
    user = await bot.get_me()
    rt.log.info("? Bot [%s] is starting...", user.full_name)

    await raito.register_commands(bot)

    yield

    rt.log.info("? Bye!")

Роутеры

Кастомные роутеры raito.Router, наследуются от aiogram.Router. Они нужны для работы с приоритетами авто-загрузки и отключением этой самой авто-загрузки.

from raito import Router

router = Router(name="start", priority=9999)
router = Router(name="debug", autoload=False)

Чем выше приоритет, тем раньше загрузится роутер, это полезно для ивентов со сложными фильтрациями. Как пример, вы можете создать хэндлер @router.on_message() с приоритетом -9999, который будет ловить все оставшиеся сообщения, которые не прошли проверку другими хэндлерами.

Если вы выключите авто-подключение роутера при старте бота (autoload=False), для его использования вам потребуется вручную включить его в Telegram чате: .rt load debug

Загрузка роутеров в .rt load/unload/reload <name> командах происходит по их параметру name. Вы можете вывести список всех найденных роутеров в боте через .rt routers

И помимо этого, роутеры содержат свои кастомные ивенты.
Есть 2 способа регистрации этих ивентов, если по каким-либо причинам вы не хотите использовать raito.Router:

1. Через raito.rt

   from aiogram import Router
   
   from raito import rt
   
   router = Router() # aiogram.Router
   
   
   @rt.lifespan(router)
   async def lifespan(): ...
   
   @rt.on_pagination(router, "unique_name")
   async def on_pagination(): ...
   

2. Через raito.Router (Рекомендую)

   from raito import Router
   
   router = Router() # raito.Router
   
   
   @router.lifespan()
   async def lifespan(): ...
   
   @router.on_pagination("unique_name")
   async def on_pagination(): ...
   
   @router.on_command_signature_error()
   async def on_command_signature_error(): ...
   

Команды

Командами называются on_message ивенты, которые вызываются через слеш-префикс ( /start ). Команды можно зарегистрировать в слэш-меню через @BotFather или через bot.set_my_commands.

Это неудобно делать вручную, и конечно, хочется это автоматизировать.

Здесь Вам и поможет Raito. Благодаря тому, что Raito полностью контролирует подключение роутеров в диспетчер, он также видит и все команды.

Для добавления в слэш-меню требуются description и вы можете их указать для каждой команды через декоратор @rt.description():

from aiogram import Router, filters
from aiogram.utils.i18n import lazy_gettext as __

from raito import rt

router = Router(name="admin")


@router.message(filters.CommandStart())
@rt.description("Start command")
async def start(): ...

@router.message(filters.Command("help"))
@rt.description(__("Help description with i18n"))
async def help(): ...

А чтобы команды не попали в слэш-меню, скройте их через @rt.hidden:

from aiogram import filters

from raito import Router, rt

router = Router(name="debug")


@router.message(filters.Command("debug"))
@rt.hidden
async def debug(): ...

Есть еще @rt.params, но я предлагаю вам изучить его самостоятельно.

Роли

RBAC (Role Based Access Control) это классика. Вот как удобно реализована система доступов в Raito:

from aiogram import Router, filters

from raito.plugins.roles import ADMINISTRATOR, MODERATOR, OWNER

router = Router(name="moderation")


@router.message(filters.Command("ban"), OWNER | ADMINISTRATOR | MODERATOR)
async def ban(): ...

A | B означает "одна из двух ролей" (OR операция). У пользователя может быть только одна роль. В этом хэндлере вы проверяете, является ли он владельцем бота, администратором или модератором.

Вы можете выдавать роли через команду .rt roles и просматривать список всех привилегированных пользователей через .rt staff

По умолчанию используется MemoryStorage, это значит, что все выданные роли сбросятся после перезапуска бота. Чтобы этого не происходило, можете использовать RedisStorage.

Но тащить целый Redis в проект ради хранения ролей неэффективно, поэтому я создал JSONStorage, SQLiteStorage и PostgreSQLStorage.

Эти 3 хранилища, к слову, полностью реализуют абстрактный класс aiogram.BaseStorage, поэтому вы сможете использовать эти хранилище даже в самом aiogram.Dispatcher(storage=...). Все states и data из aiogram FSM будут храниться в соответствующих хранилищах. Но это, конечно, экзотический способ, решайте сами.

Подключение внешнего хранилища для ролей происходит вот так:

from raito import Raito
from raito.utils.storages import get_sqlite_storage

SQLiteStorage = get_sqlite_storage()
storage = SQLiteStorage("sqlite+aiosqlite:///bot.db")

raito = Raito(dispatcher, "src/handlers", storage=storage)

А создать свою роль еще проще:

from raito.plugins.roles.constraint import RoleConstraint
from raito.plugins.roles.filter import RoleFilter

DUDE = RoleConstraint(
    RoleFilter(slug="dude", name="Dude", description="Just a dude", emoji="?")
)

Подробнее о том, как переписать под себя бизнес-логику ролей в документации.

Клавиатуры

Одно из самых главных преимуществ ботов - кнопки. И не менее важно, комфортно ими управлять.

Статичный способ

from aiogram import Router, filters, types

from raito import rt

router = Router(name="start")


@rt.keyboard.static(inline=False)
def start_markup():
    return [
        ["? Throw a ball"],  # ряд с одной кнопкой
        [["? FAQ"], ["? Leaderboard"]],  # ряд с двумя кнопками
    ]


@router.message(filters.CommandStart())
async def start(message: types.Message):
    await message.answer(text="Welcome!", reply_markup=start_markup())

Динамичный способ

from aiogram import Router, filters, types
from aiogram.utils.keyboard import InlineKeyboardBuilder

from raito import rt

router = Router(name="info")

# 1, 2 - означает 1 кнопка в первом ряду, 2 кнопки во втором ряду
@rt.keyboard.dynamic(1, 2, inline=True)
def links_markup(builder: InlineKeyboardBuilder, privacy_url: str, tos_url: str):
    builder.button(text="? Support", callback_data="support")
    builder.button(text="? Privacy", url=privacy_url)
    builder.button(text="? TOS", url=tos_url)


@router.message(filters.Command("info"))
async def info(message: types.Message):
    await message.answer(
        text="Information",
        reply_markup=links_markup(
            privacy_url="https://example.com/privacy",
            tos_url="https://example.com/tos",
        ),
    )

Пагинация

Есть множество библиотек для пагинации, практически все из них используют внешнее хранилище. Мне не нравится эта концепция.

Telegram дает возможность хранить 64 символа в callback_data, этим я и воспользовался: rt_p:<mode>:<name>:<page>:<total>:<limit>

Вот как это работает:

from aiogram import Bot, filters, types

from raito import Raito, Router
from raito.plugins.pagination import InlinePaginator, PaginationMode

router = Router(name="pagination")


@router.message(filters.Command("pagination"))
async def pagination(message: types.Message, raito: Raito, bot: Bot):
    if not message.from_user:
        return

    await raito.paginate(
        "my_pagination_name",
        chat_id=message.chat.id,
        bot=bot,
        from_user=message.from_user,
        total_pages=10,
        limit=5,
        mode=PaginationMode.INLINE,
    )


@router.on_pagination("my_pagination_name")
async def on_pagination(
    query: types.CallbackQuery,
    paginator: InlinePaginator,
    offset: int,
    limit: int,
):
    buttons = [
        types.InlineKeyboardButton(text=str(i), callback_data=f"button_{i}")
        for i in range(offset, offset + limit)
    ]
    await paginator.answer("Button list:", buttons=buttons)

Режимы пагинации:

• INLINE: Необязательный текст с inline кнопками

• TEXT: Только текст с навигацией

• PHOTO: Медиа-контент с необязательной подписью (caption <2048 символов)

• LIST: Грубо говоря, такой же TEXT, но отформатированный как список строк, соединенных разделителем.

• RICH: Новый формат rich-сообщений появившийся в недавнем обновлении Telegram. Требуется aiogram>=3.30.0.

Сцены

Несмотря на то, что FSM реализованы хорошо, им не хватает валидации данных и навигации.

aiogram предлагает FSM Scene Wizard как альтернативу, но они выполнены достаточно сложно и громоздко, до такой степени, что их не использовал никто в серьезных проектах.

Пример продвинутого использования сценариев анкетирования:

from aiogram import F, filters
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import Message
from sqlalchemy.ext.asyncio import AsyncSession

from raito import Router
from raito.plugins.scenes import Scene, SceneData

router = Router(name="moderation")


class MuteData(SceneData):
    username: str | None = None
    minutes: int | None = None


class MuteStates(StatesGroup):
    username = State()
    minutes = State()


mute = router.scene(MuteStates, data=MuteData)


@mute.on_message.enter(filters.Command("mute"))
async def start(message: Message, scene: Scene[MuteData]) -> None:
    await message.answer("Enter username:")
    await scene.next()


@mute.on_message(MuteStates.username, F.text)
async def set_username(message: Message, scene: Scene[MuteData]) -> None:
    if not (message.text or "").startswith("@"):
        await message.answer("⚠️ Enter an @username")
        return await scene.retry()

    scene.data.username = message.text
    await message.answer("Enter duration in minutes:")
    await scene.next()


@mute.on_message(MuteStates.minutes, F.text)
async def set_minutes(
    message: Message,
    scene: Scene[MuteData],
    session: AsyncSession,
) -> None:
    if not (message.text or "").isdigit() or int(message.text) <= 0:
        await message.answer("⚠️ Enter a positive whole number")
        return await scene.retry()

    scene.data.minutes = int(message.text)
    await mute_user(session, scene.data.username, scene.data.minutes)
    await message.answer("✅ User muted")
    await scene.finish()

Диалоги

Диалоги это упрощенная версия FSM.

Отмечу сразу, что Диалоги использовать в высоконагруженных ботах не рекомендуется, в связи с тем, что они используют asyncio.Future и удерживают пул сессий. Вместо этого, используйте Сцены. Но если Ваш бот не держит высокий DAU или вы делаете MVP, Диалоги вам подойдут.

from aiogram import F, Router, filters
from aiogram.fsm.context import FSMContext
from aiogram.types import Message

from raito import Raito

router = Router(name="mute")


@router.message(filters.Command("mute"))
async def mute(message: Message, raito: Raito, state: FSMContext) -> None:
    await message.answer("Enter username:")
    user = await raito.wait_for(state, F.text.regexp(r"@[\w]+"))

    await message.answer("Enter duration (in minutes):")
    duration = await raito.wait_for(state, F.text.isdigit())

    while not duration.number or duration.number < 0:
        await message.answer("⚠️ Duration cannot be negative")
        duration = await duration.retry()

    await message.answer(f"✅ {user.text} will be muted for {duration.number} minutes")

Вам может показаться, что все происходит внутри одного хэндлера, но под капотом wait_for создают Future (аналог Promise в JS) и заносят его в регистр. Затем миддлварь отслеживает все входящие сообщения, прогоняет их через фильтры (второй аргумент wait_for) и передает/резолвит в wait_for этот event object.

Благодаря этому у разработчиков есть самый простой на текущий момент способ запросить ответ пользователя. Но и без побочных эффектов никуда.

В aiogram есть возможность включить Dispatcher(events_isolation=SimpleEventIsolation()), это режим блокировки, когда один пользователь может вызвать только один хэндлер за раз. Вот только из-за wait_for хэндлер "не отпускает пользователя" никогда.

Простыми словами: Диалоги не работают с SimpleEventIsolation.

Троттлинг

И под конец, ограничение вызовов хэндлеров.

Для каждого хэндлера отдельно:

from aiogram import Router, filters, types

from raito import rt

router = Router(name="export")


@router.message(filters.Command("export"))
@rt.limiter(rate_limit=5.0, mode="user")
async def export(message: types.Message) -> None: ...

Глобальный для всех хэндлеров:

from aiogram import Bot

from raito import Router, rt
from raito.core.raito import Raito

router = Router(name="lifespan")


@router.lifespan()
async def lifespan(bot: Bot, raito: Raito):
    user = await bot.get_me()
    rt.log.info("? Bot [%s] is starting...", user.full_name)

    await raito.register_commands(bot)

    # ↓↓↓
    raito.add_throttling(rate_limit=0.5, mode="user", max_size=100_000)

    yield

    rt.log.info("? Bye!")

Режимы:

Заключение

Спасибо, что дочитали до этого момента.

Я не стал расписывать досконально про все функции Raito, иначе статья очень сильно растянулась бы. Поэтому предлагаю Вам попробовать плагин на практике:

https://github.com/Aidenable/Raito

Помогите звездочкой, чтобы о проекте узнали больше людей...