Aiogram 3: Первые Шаги

https://habr.com/ru/companies/amvera/articles/820527/

Создание Бота через BotFather

Для начала перейдем в BotFather (специальный телеграмм бот для генерации друих ботов) и создадим нашего бота:

• Жмем на «Старт»
• Вводим команду /newbot
• Указываем имя бота (можно на русском, можно будет изменить)
• Указываем логин бота (логин должен быть уникальным и содержать приписку BOT в любом регистре, после создания изменить будет нельзя)
• Копируем токен бота

Весь путь на одном скрине. Токен сохраните и никому не показывайте!

Структура Проекта

Предлагаю проверенный временем формат стартового шаблона. Шаблон прошел проверку множеством проектов. Эта структура удобна в поддержке и легко расширяется. Вот как она выглядит:

- db_handler/
  - __init__.py
  - db_class.py
- handlers/
  - __init__.py
  - start.py
- keyboards/
  - __init__.py
  - all_keyboards.py
- work_time/
  - __init__.py
  - time_func.py
- utils/
  - __init__.py
  - my_utils.py
- filters/
  - __init__.py
  - is_admin.py
- middlewares/
  - __init__.py
  - check_sub.py
- .env
- aiogram_run.py
- create_bot.py
- requirements.txt
- run.py

Вся структура на скрине.

Теперь давайте разбираться со структурой и с наполнением каждого файла.

Начнем с файла .env:

TOKEN=720000:AAG2000Aa70p6eotkKiMKJD_nwosSAfvg
ADMINS=00000000,000000001
PG_LINK=postgresql://USER_LOGIN:USER_PASSWORD@HOST_API:PORT/NAME_BD

• TOKEN: Токен, выданный BotFather
• ADMINS: Список телеграмм ID администраторов (можно хранить и в базе данных, но на старте так удобнее)
• PG_LINK: Ссылка подключения к базе данных PostgreSQL

По поводу PostgreSQL. Я разработал собственный класс на основе asyncpg, который позволяет универсально работать как с Telegram-ботами, так и с любыми другими проектами, где требуется асинхронное взаимодействие с базой данных.

В сегодняшней статье я не буду рассматривать код моего класса, но если вам будет интересно, мы обязательно обсудим его в следующих публикациях. Если вы новичок в работе с PostgreSQL, рекомендую ознакомиться с моей статьей, в которой я подробно рассказал, как за 5 минут развернуть Docker-контейнер с PostgreSQL на своем VPS сервере. Надеюсь, что эта статья окажется для вас полезной и вы сможете вывести ее из кармической ямы. https://habr.com/ru/articles/820129/

Файл requirements.txt:

asyncpg
aiogram
APScheduler
python-decouple

• asyncpg: Библиотека для асинхронного взаимодействия с PostgreSQL
• aiogram: Библиотека для создания телеграм-ботов
• APScheduler: Библиотека для планирования задач
• python-decouple: Библиотека для работы с .env файлами

Устанавливаем зависимости:

pip install -r requirements.txt

Пакет db_handler

В этом пакете я всегда размещаю файл с универсальным хендлером для работы с PostgreSQL. Я использую его во всех своих проектах, будь то Telegram-боты или любые другие приложения, где требуется база данных.

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

Пакет handlers

Здесь я размещаю хендлеры для работы с ботом: админ-панель, пользовательскую часть, анкеты и т.д. В примере это отдельные файлы, но в крупных проектах они структурируются в отдельные пакеты под разные задачи.

С появлением роутинга в версии 3 стало ещё проще: теперь можно назначить каждому хендлеру свой роутер, не перемещая диспетчер. Сегодня я покажу, как это делать.

Пакет keyboards

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

Пакет work_time

Здесь я прописываю функции, которые должны работать по расписанию с использованием модуля APScheduler. Сегодня мы эту тему затрагивать не будем, но знайте, что такая возможность существует.

Пакет utils

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

Пакет middlewares

Middleware в aiogram — это промежуточный слой, который вызывается автоматически после получения запроса и перед его обработкой сервером. Они могут быть полезны для выполнения различных задач, таких как:

• Логгирование запросов
• Добавление полезной нагрузки

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

Пакет filters

Фильтры в aiogram позволяют ограничивать обработку определённых типов сообщений или команд, что помогает создать более точную и гибкую логику обработки запросов. Например, с помощью фильтров можно настроить бота так, чтобы он реагировал только на текстовые сообщения, команды от конкретных пользователей или проверял, является ли пользователь администратором.

Файл run.py

В этом файле прописываю инструкции для systemd, чтобы быстро развернуть бота на сервере в режиме автозапуска. Этой теме планирую посвятить отдельную статью, сегодня касаться не будем.

Файл create_bot.py

Создаем файл create_bot.py:

import logging
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.fsm.storage.memory import MemoryStorage
from decouple import config
from apscheduler.schedulers.asyncio import AsyncIOScheduler

from db_handler.db_class import PostgresHandler

pg_db = PostgresHandler(config('PG_LINK'))
scheduler = AsyncIOScheduler(timezone='Europe/Moscow')
admins = [int(admin_id) for admin_id in config('ADMINS').split(',')]

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

bot = Bot(token=config('TOKEN'), default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher(storage=MemoryStorage())

Разбор импортов и инициализации:

• Импортируем необходимые библиотеки и модули.
• Создаем объект для работы с базой данных PostgreSQL (на основе моего класса, пока можно закомментировать).
• Создаем планировщик задач.
• Настраиваем список администраторов.
• Настраиваем логирование.
• Инициализируем бота и диспетчера.

Разбор импортов

• import logging: Импортируем библиотеку для логирования, чтобы записывать события и ошибки в процессе работы бота.
• from aiogram import Bot, Dispatcher: Импортируем классы Bot и Dispatcher из библиотеки aiogram, которые необходимы для создания и управления ботом.
• from aiogram.fsm.storage.memory import MemoryStorage: Импортируем класс MemoryStorage для хранения состояний конечного автомата (FSM) в памяти.
• from decouple import config: Импортируем функцию config из библиотеки python-decouple для загрузки переменных окружения из файла .env.
• from apscheduler.schedulers.asyncio import AsyncIOScheduler: Импортируем класс AsyncIOScheduler из библиотеки APScheduler для планирования задач (например, выполнение скриптов по времени).
• from db_handler.db_class import PostgresHandler: Импортируем твой кастомный класс PostgresHandler для работы с базой данных Postgres.

Инициализация объектов

pg_db = PostgresHandler(config('PG_LINK'))

pg_db = PostgresHandler(config('PG_LINK')): Создаем объект класса PostgresHandler для работы с базой данных. Строка подключения к базе данных загружается из переменной окружения PG_LINK. Завязана инициализация на мой класс о котором поговорим в следующий раз.

scheduler = AsyncIOScheduler(timezone='Europe/Moscow')

scheduler = AsyncIOScheduler(timezone='Europe/Moscow'): Создаем объект AsyncIOScheduler для планирования и выполнения задач по времени. Устанавливаем часовой пояс на Europe/Moscow.

Настройка администраторов

admins = [int(admin_id) for admin_id in config('ADMINS').split(',')]

admins = [int(admin_id) for admin_id in config('ADMINS').split(',')]: Создаем список ID администраторов бота. Загружаем строку с ID администраторов из переменной окружения ADMINS, разделяем её по запятым и преобразуем каждый элемент в целое число.

Настройка логирования

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') 
logger = logging.getLogger(__name__)

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'): Настраиваем базовое логирование с уровнем INFO, чтобы записывать важные сообщения. Устанавливаем формат логов, включающий время, имя логгера и уровень сообщения.

logger = logging.getLogger(__name__): Создаем логгер с именем текущего модуля, чтобы записывать лог-сообщения.

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

bot = Bot(token=config('TOKEN'), default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher(storage=MemoryStorage())

bot = Bot(token=config('TOKEN'),default=DefaultBotProperties(parse_mode=ParseMode.HTML)): Создаем объект Bot с токеном, загруженным из переменной окружения TOKEN. По умолчанию прописал, чтоб бот корректно вопринимал HTML теги для форматирования текста (заслуживает отдельного обсуждения).

Dispatcher

dp = Dispatcher(storage=MemoryStorage()) в данном контексте выполняет несколько ключевых задач в рамках создания Telegram-бота с использованием библиотеки aiogram:

1. Создание диспетчера:

   • Dispatcher - это основной объект, отвечающий за обработку входящих сообщений и других обновлений, поступающих от Telegram. Именно через диспетчер проходят все сообщения и команды, отправляемые пользователями бота.

2. Настройка хранилища состояния:

   • storage=MemoryStorage() указывает, что для хранения состояния конечных автоматов (FSM) используется память (RAM). Это значит, что состояния пользователей будут храниться в оперативной памяти (тема большой отдельной статьи).

   • FSM (Finite State Machine) используется для управления состояниями диалога с пользователем. Это позволяет боту "помнить" текущий шаг разговора и реагировать на действия пользователя в соответствии с текущим состоянием.

Файл aiogram_run.py

Создаем файл aiogram_run.py для запуска бота:

import asyncio
from create_bot import bot, dp, scheduler
from handlers.start import start_router
# from work_time.time_func import send_time_msg

async def main():
    # scheduler.add_job(send_time_msg, 'interval', seconds=10)
    # scheduler.start()
    dp.include_router(start_router)
    await bot.delete_webhook(drop_pending_updates=True)
    await dp.start_polling(bot)

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

Разбор кода:

• Импортируем необходимые модули и функции.
• Определяем основную асинхронную функцию main.
• Настраиваем и запускаем бота в режиме опроса.

Так как мы с вами не писали work_time.time_func код данный импорт, как и использование в главной функции пока комментируем. К данному вопросу вернемся позже.

Основная асинхронная функция

async def main():
    # scheduler.add_job(send_time_msg, 'interval', seconds=10)
    # scheduler.start()
    dp.include_router(start_router)
    await bot.delete_webhook(drop_pending_updates=True)
    await dp.start_polling(bot)

• async def main():Определяем основную асинхронную функцию main, которая будет запускаться при старте бота.
• scheduler.add_job(send_time_msg, 'interval', seconds=10): Добавляем задачу в планировщик scheduler. Задача send_time_msg будет выполняться каждые 10 секунд.
• scheduler.start(): Запускаем планировщик задач, чтобы он начал выполнять добавленные задачи по расписанию.
• dp.include_router(start_router): Добавляем роутер start_router в диспетчер dp. Это позволяет диспетчеру знать о всех обработчиках команд, которые определены в start_router. Данный роутер мы напишем немного позже, но включение роутера можем сделать уже сейчас.
• await bot.delete_webhook(drop_pending_updates=True): Несмотря на то, что мы работаем через метод лонг поллинга, данная строка так же будет корректной. В тройке это аналог записи: skip_updates=True из более старых версий aiogram
• await dp.start_polling(bot): Запускаем бота в режиме опроса (polling). Бот начинает непрерывно запрашивать обновления с сервера Telegram и обрабатывать их (о том как писать aiogram 3 бота через технологию вебхуков в связке с FastApi я писал в одной из своих прошлых публикаций).

Файл handlers/start.py

Создаем файл handlers/start.py с нашим первым хендлером:

from aiogram import Router, F
from aiogram.filters import CommandStart, Command
from aiogram.types import Message

start_router = Router()

@start_router.message(CommandStart())
async def cmd_start(message: Message):
    await message.answer('Запуск сообщения по команде /start используя фильтр CommandStart()')

@start_router.message(Command('start_2'))
async def cmd_start_2(message: Message):
    await message.answer('Запуск сообщения по команде /start_2 используя фильтр Command()')

@start_router.message(F.text == '/start_3')
async def cmd_start_3(message: Message):
    await message.answer('Запуск сообщения по команде /start_3 используя магический фильтр F.text!')

Импорты:

Из библиотеки aiogram мы импортировали Router и F, также известный как "магический фильтр".

• Router используется для удобного масштабирования проекта. Благодаря ему, мы можем отказаться от необходимости импортировать Dispatcher в каждом хендлере. Вместо этого роутеры берут на себя эту роль, упрощая структуру кода и улучшая его читаемость.

• F, или "магический фильтр", позволяет "на лету" фильтровать входящие события и выдавать нужные результаты. По сравнению со старой версией aiogram, использование F сокращает код в полтора-два раза в некоторых случаях.

• CommandStart и Command – это встроенные фильтры. CommandStart срабатывает на команду /start, а Command активируется при любой команде, переданной аргументом. Примеры использования этих фильтров вы увидите в коде ниже (на разборе функций).

• Message используется для аннотации типов. Это позволяет IDE, такой как PyCharm, лучше понимать, с какими данными мы работаем, и предоставлять более точные подсказки. Объект Message содержит множество полезной информации:

  • telegram_id
  • user_name
  • first_name
  • last_name

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

Доступ к данным осуществляется через точку. Пример: message.text (получаем текст сообщения).

Рассмотрим каждую функцию в отдельности

@start_router.message(CommandStart())
async def cmd_start(message: Message):
    await message.answer('Запуск сообщения по команде /start используя фильтр CommandStart()')

Декоратор @start_router.message(CommandStart())

Эта функция использует декоратор @start_router.message с фильтром CommandStart(). Такой подход избавляет нас от необходимости вручную регистрировать каждую функцию в файле запуска бота (например, в register_handler). Благодаря такой реализации, разработка становится проще и чище.

Аннотация типов для message

Мы передали в декоратор параметр message с аннотацией типа Message. Это позволяет выполнять await message.answer, что аналогично использованию await bot.send_message, но без необходимости передавать chat_id. Бот автоматически понимает, кому нужно отправить сообщение.

В данной функции бот отправляет заранее подготовленное сообщение при команде /start. Это стало возможным благодаря фильтру CommandStart().

@start_router.message(Command('start_2'))
async def cmd_start_2(message: Message):
    await message.answer('Запуск сообщения по команде /start_2 используя фильтр Command()')

Эта функция аналогична предыдущей, но использует фильтр Command(). В этом случае мы явно указываем команду, на которую должен реагировать бот /start_2. Остальная логика остаётся такой же.

@start_router.message(F.text == '/start_3')
async def cmd_start_3(message: Message):
    await message.answer('Запуск сообщения по команде /start_3 используя магический фильтр F.text!')

Здесь мы применили фильтр F.text, который позволяет фильтровать сообщения по содержимому текста. В данном примере бот реагирует на текст /start_3.

Интеграция роутера

Не забудьте включить наш роутер в основном файле бота:

dp.include_router(start_router)

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

Запускаем бота и смотрим что у нас получилось:

Смотрим на логи.

Прекрасно, что бот успешно запущен и работает в режиме поллинга! Это значит, что бот периодически проверяет сервер Telegram на наличие новых сообщений.

При запуске бот вывел информацию о себе:

• Имя бота
• Логин бота
• ID бота

Переходим в бота:

Конечно, пока внешний вид бота может показаться немного простым, но это легко исправить. Мы всегда можем улучшить его, добавив логотип, описание и приветственное фото.

А теперь нажимаем на «ЗАПУСТИТЬ» и смотрим, что у нас получилось:

Мы видим, что бот отреагировал на команду /start именно так, как мы это задумали. Теперь давайте попробуем выполнить команду /start_2 и /start_3:

И остальные команды отработали безупречно, что подтверждает функциональность наших фильтров и роутера.