Как просто создать aiogram 3.x бота на вебхуках (webhook)?

Приветствую, Хабр! Меня зовут Алексей, и я опытный 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)

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