SavePearlHarbor

Облегчи себе разработку ботов

—

от автора

admin

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

Боты простые. У них есть кнопки, они принимают сообщения и отвечают сообщениями.

Но даже в бот-деве не все идеально.

Мы все используем 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

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

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

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

Основы

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

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

# __main__.pyimport asynciofrom aiogram import Bot, Dispatcherfrom raito import Raitoasync 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.pyfrom aiogram import Botfrom raito import Router, rtfrom raito.core.raito import Raitorouter = 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 Routerrouter = 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 Routerfrom raito import rtrouter = 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 Routerrouter = 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, filtersfrom aiogram.utils.i18n import lazy_gettext as __from raito import rtrouter = 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 filtersfrom raito import Router, rtrouter = Router(name="debug")@router.message(filters.Command("debug"))@rt.hiddenasync def debug(): ...

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

Роли

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

from aiogram import Router, filtersfrom raito.plugins.roles import ADMINISTRATOR, MODERATOR, OWNERrouter = 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 Raitofrom raito.utils.storages import get_sqlite_storageSQLiteStorage = get_sqlite_storage()storage = SQLiteStorage("sqlite+aiosqlite:///bot.db")raito = Raito(dispatcher, "src/handlers", storage=storage)

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

from raito.plugins.roles.constraint import RoleConstraintfrom raito.plugins.roles.filter import RoleFilterDUDE = RoleConstraint(    RoleFilter(slug="dude", name="Dude", description="Just a dude", emoji="😎"))

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

Клавиатуры

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

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

from aiogram import Router, filters, typesfrom raito import rtrouter = 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, typesfrom aiogram.utils.keyboard import InlineKeyboardBuilderfrom raito import rtrouter = 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, typesfrom raito import Raito, Routerfrom raito.plugins.pagination import InlinePaginator, PaginationModerouter = 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, filtersfrom aiogram.fsm.state import State, StatesGroupfrom aiogram.types import Messagefrom sqlalchemy.ext.asyncio import AsyncSessionfrom raito import Routerfrom raito.plugins.scenes import Scene, SceneDatarouter = Router(name="moderation")class MuteData(SceneData):    username: str | None = None    minutes: int | None = Noneclass 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, filtersfrom aiogram.fsm.context import FSMContextfrom aiogram.types import Messagefrom raito import Raitorouter = 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, typesfrom raito import rtrouter = 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 Botfrom raito import Router, rtfrom raito.core.raito import Raitorouter = 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!")

Режимы:

Режим

Кулдаун

«user»

Для одного пользователя (во всех чатах)

«chat»

Для одного чата (общий для всех участников)

«bot»

Для всех сразу (единый глобальный кулдаун)

Заключение

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

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

https://github.com/Aidenable/Raito

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

ссылка на оригинал статьи https://habr.com/ru/articles/1063000/