Приветствую, Хабр! Меня зовут Алексей, и я опытный Python-разработчик с многолетним стажем. Как и многие другие, я начинал с создания телеграм-ботов, используя метод лонг поллинга. Однако, передо мной встала задача реализации бота через вебхуки, и я решил поделиться своим опытом с вами.

На сегодняшний день я уже хорошо знаком с FastAPI, умею настраивать серверы и поднимать NGINX с защищённым сертификатом HTTPS. Для этой статьи мы будем считать, что вы тоже имеете эти навыки. Если будет необходимость, я с удовольствием опишу, как создать базовый шаблон FastAPI и настроить VPS сервер, но сейчас будем считать, что всё уже настроено.

Итак, сервер у нас готов, и теперь мы приступим к созданию бота на aiogram 3.x с использованием вебхуков.

Установка и настройка

Для начала установим последнюю версию aiogram (на момент написания это aiogram 3.7.x):

pip install aiogram

Супер. Теперь давайте настроим файл бота. Усложнять сейчас не будем и всё пропишем в одном файле, назовём его bot.py.

Импорты

Для начала выполним следующие импорты:

import logging
from aiogram.client.default import DefaultBotProperties
from aiogram import Bot, Dispatcher, F
from aiogram.types import Message, Update
from aiogram.filters import CommandStart
from aiogram.enums import ParseMode
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.requests import Request
import uvicorn
from contextlib import asynccontextmanager

Инициализация бота и диспетчера

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

bot = Bot(token="ВАШ_ТОКЕН", default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher()

Запись default=DefaultBotProperties(parse_mode=ParseMode.HTML) позволяет боту читать HTML теги в сообщениях (например, <b></b>).

Настройка вебхуков

Самое важное:

@asynccontextmanager
async def lifespan(app: FastAPI):
    url_webhook = ССЫЛКА НА САЙТ + ПУТЬ К ВЕБХУКУ' (пример: https://example.ru/webhook)
    await bot.set_webhook(url=url_webhook,
                          allowed_updates=dp.resolve_used_update_types(),
                          drop_pending_updates=True)
    yield
    await bot.delete_webhook()

Конечно, давайте разберем функцию lifespan более подробно.

Что делает эта функция?

Функция lifespan отвечает за жизненный цикл (lifespan) вашего FastAPI приложения. Она используется для выполнения действий, которые нужно сделать до запуска сервера и после его остановки. В данном случае, она устанавливает и удаляет вебхук для вашего телеграм-бота.

  1. Декоратор @asynccontextmanager:

    • Этот декоратор используется для создания асинхронного контекстного менеджера. Контекстные менеджеры позволяют управлять ресурсами, которые нужно инициализировать и освобождать (например, подключение к базе данных, сетевые подключения и т.д.).

  2. Параметр app:

    • Это ваш FastAPI объект, который используется для настройки приложения и добавления маршрутов.

  3. Переменная url_webhook:

    • url_webhook формируется из базового URL вашего сайта и пути к вашему вебхуку. Это URL, на который Telegram будет отправлять обновления для вашего бота. Пример: 'https://example.ru/webhook'.

Инициализация FastAPI

Теперь после этих настроек можем приступать к самому FastAPI приложению.

app = FastAPI(lifespan=lifespan)
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
  • app = FastAPI(lifespan=lifespan):

    • Здесь мы создаем экземпляр приложения FastAPI.

    • Параметр lifespan=lifespan передается в конструктор FastAPI. Это означает, что FastAPI будет использовать нашу функцию lifespan для управления жизненным циклом приложения.

    • Функция lifespan определяет действия, которые выполняются при запуске приложения (установка вебхука) и при его остановке (удаление вебхука).

  • app.mount("/static", StaticFiles(directory="static"), name="static"):

    • Метод mount используется для "монтирования" других приложений или ресурсов к маршрутам FastAPI.

    • В данном случае, мы монтируем каталог со статическими файлами.

    • "/static": Это путь, по которому будут доступны статические файлы. Например, файл static/example.png будет доступен по URL http://your-domain/static/example.png.

    • StaticFiles(directory="static"): Указывает, что каталог static в корневом каталоге проекта будет использоваться для хранения и сервировки статических файлов.

    • name="static": Это имя, под которым ресурс будет зарегистрирован в приложении. Это полезно для внутренней идентификации ресурса.

  • templates = Jinja2Templates(directory="templates"):

    • Здесь мы создаем объект Jinja2Templates, который будет использоваться для рендеринга HTML-шаблонов.

    • directory="templates": Указывает, что шаблоны будут находиться в каталоге templates. Например, если у вас есть файл index.html в каталоге templates, вы сможете рендерить его при обработке запросов.

Зачем это нужно?

  1. Управление жизненным циклом:

    • lifespan позволяет управлять действиями при запуске и остановке приложения, что критично для задач, требующих инициализации и очистки, таких как установка и удаление вебхуков.

  2. Обслуживание статических файлов:

    • Монтирование статических файлов позволяет вашему приложению обслуживать CSS, JavaScript, изображения и другие файлы напрямую из указанного каталога. Это удобно для поддержки фронтенда и статики.

  3. Шаблоны:

    • Jinja2Templates предоставляет мощный механизм для рендеринга HTML-шаблонов с данными из вашего приложения, что позволяет создавать динамические веб-страницы.

Обработчики запросов

Функция для запуска index.html по корневому пути (если вам нужно):

@app.get("/", response_class=HTMLResponse)
async def read_root(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

Функция, которая привязывает вебхук:

@app.post("/webhook")
async def webhook(request: Request) -> None:
    update = Update.model_validate(await request.json(), context={"bot": bot})
    await dp.feed_update(bot, update)
  • @app.post("/webhook"): Это декоратор, который говорит FastAPI, что эта функция будет обрабатывать POST-запросы по маршруту /webhook.

  • /webhook: Это путь, по которому Telegram будет отправлять обновления для вашего бота. Вам нужно указать этот URL в настройках вебхука вашего бота.

Полный файл с запуском

Ну и давайте теперь к примеру полного файла бота. Думаю будет все понятно после разбора:

import logging
from aiogram.client.default import DefaultBotProperties
from aiogram import Bot, Dispatcher, F
from aiogram.types import Message, Update
from aiogram.filters import CommandStart
from aiogram.enums import ParseMode
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.requests import Request
import uvicorn
from contextlib import asynccontextmanager

bot = Bot(token='7414957579:AAEYqGD3OTcp4DxfHud6NOJJU8zYlWeIHvU',
          default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher()


@asynccontextmanager
async def lifespan(app: FastAPI):
    await bot.set_webhook(url="ССЫЛКА С ВЕБХУКОМ",
                          allowed_updates=dp.resolve_used_update_types(),
                          drop_pending_updates=True)
    yield
    await bot.delete_webhook()


app = FastAPI(lifespan=lifespan)
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")


@dp.message(CommandStart())
async def start(message: Message) -> None:
    await message.answer('Привет!')


@app.get("/", response_class=HTMLResponse)
async def read_root(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})


@app.post("/webhook")
async def webhook(request: Request) -> None:
    update = Update.model_validate(await request.json(), context={"bot": bot})
    await dp.feed_update(bot, update)


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format=u'%(filename)s:%(lineno)d #%(levelname)-8s [%(asctime)s] - %(name)s - %(message)s',
    )

    uvicorn.run(app, host="0.0.0.0", port=5000)

Надеюсь, что это было полезно. Если у вас есть вопросы или замечания, оставляйте их в комментариях. Буду рад помочь и учесть ваши пожелания в будущих публикациях. Спасибо за внимание!

Комментарии (14)


  1. Pol1mus
    06.06.2024 13:34
    +1

    Миллион таких же инструкций видел. У всех одна и та же проблема, куча лишнего непонятного и чего то явно не хватает.


    1. yakvenalex Автор
      06.06.2024 13:34

      А что непонятно? Вроде максимально дотошно описывал (поэтому и "куча лишнего")


      1. Pol1mus
        06.06.2024 13:34

        Ну у тебя например не упоминаются сертификаты. И порт не такой как в других инструкциях.

        В принципе не понятно что происходит и как.


        1. yakvenalex Автор
          06.06.2024 13:34

          За сертификаты отвечает fastapi если его правильно настроить. Сегодня-завтра пост выложу по этому поводу. Так по деталям смотри в моих постах


          1. t0rr
            06.06.2024 13:34

            За сертификаты отвечает реверс-прокси, не надо пичкать этим ни аиограм, ни фастапи


  1. AikoASMR
    06.06.2024 13:34

    Там же aiohttp под капотом у aiogram если вдруг темплейт нужно отдать. Также вебхуки лучше опционально делать, неудобно с ними локально, нужно ещё ngrok какой то поднимать.


    1. yakvenalex Автор
      06.06.2024 13:34

      Нет. Не обязательно. FastApi вполне справится. Просто попробуйте за мной повторить.


  1. Artarik
    06.06.2024 13:34

    Почему-то мне кажется, что часть статьи писал чатГПТ.


  1. S0mbre
    06.06.2024 13:34

    Я заметил, что в больших ботах вебхуки работают хуже, чем обычный лонг поллинг. С хуками у меня периодически зависает приложение и затем, когда отвисает, начинает флудить сообщениями бота. Поэтому я для своих ботов юзаю лонг поллинг, бот никогда не виснет. И не надо мутить веб апп, ковыряться с сертификатами (тот же traefik) и тд


    1. yakvenalex Автор
      06.06.2024 13:34

      Однозначно лонг поллинг это проще, но оно и медленней. У меня есть боты в которых за секунду бывает 200-300 запросов. Лонг поллинг в таком случае сильно буксует. По поводу "виснет", возможно дело в настройках или сервере.


      1. S0mbre
        06.06.2024 13:34

        Ну не знаю, у меня стоял стандартный FasAPI через Uvicorn / Gunicorn, в 10 процессов. Все это в отдельном docker контейнере. Серты через Traefik. Для обычных веб приложений этот стек работает на ура. А вот бот не тянет.


      1. S0mbre
        06.06.2024 13:34

        Что медленный - не заметил. У меня в боте до 50 операций в секунду, но около 1000 пользователей. Ничего не тормозит.


    1. t0rr
      06.06.2024 13:34
      +1

      Ищите проблему в своей реализации. У меня несколько миллионов активных пользователей, генерящих тысячи rps. Поллинг столько физически не тянет, а вебхуки позволяют скейлить сервис и оперативно разгребать всё, что к нему прилетает


  1. t0rr
    06.06.2024 13:34

    Одна из немногих причин использования вебхуков вместо поллинга – возможность скейлить приложение. Подход с установкой и удалением вебхуков, показанный в примере с lifespan, превратит скейлинг в ад.

    При старте приложения можно получить информацию об установленном вебхуке. Если настройки соответствуют, то можно смело ничего не делать. А если не соответствуют - вот тогда уже выполнять настройку. Удалять вебхук при завершении приложения крайне не рекомендую: замена контейнера сварме/кубере приведёт к гарантированному удалению вебхука при работающем приложении.