Telegram-бот-магазин на Python: пошаговый гайд с оплатой, каталогом и админкой (Aiogram 3 + SQLAlchemy 2)

Друзья, приветствую! Сегодня я к вам с новым практическим проектом на Python. На этот раз мы создадим полноценного Telegram-бота для продажи цифровых товаров с базой данных, которой будем управлять через SQLAlchemy 2, админ-панелью, пользовательской частью и интегрированной оплатой через Юкассу.

Я шаг за шагом проведу вас через все этапы разработки такого бота: начиная от регистрации токена в BotFather и заканчивая деплоем готового продукта на удаленном хостинге, чтобы бот мог бесперебойно работать 24/7 без привязки к вашему компьютеру или интернет-соединению. Но обо всем по порядку.

Какие технологии мы будем использовать?

В рамках этого проекта мы будем работать исключительно с Python, так что те, кто не любит JavaScript, сегодня могут расслабиться. Мы используем следующие технологии:

• Aiogram 3 — лучший асинхронный фреймворк на Python для разработки Telegram-ботов.

• SQLAlchemy 2 — мощный ORM на Python, поддерживающий асинхронное взаимодействие со всеми табличными базами данных. Мы будем работать с SQLite, но написанный код также подойдет для других баз, например PostgreSQL.

• Aiosqlite — асинхронный движок для взаимодействия с SQLite через SQLAlchemy 2.

• Pydantic 2 — для валидации данных и работы с настройками.

• Alembic — для автоматизации работы со структурами таблиц (миграции).

Важно: дисклеймер!

Мы разрабатываем магазин цифровых товаров, и я покажу «классическое» подключение оплаты в боте через BotFather. Однако с 12 июня 2024 года появилось обязательство для Telegram-ботов по продаже цифровых товаров использовать систему оплаты «Telegram Stars» (звезды).

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

В будущих публикациях я расскажу, как работать с «звездами» в РФ. Но пока надеемся, что блокировки нас не коснутся, и продолжаем.

Пошаговый план

В рамках проекта мы выполним следующие этапы:

1. Подготовка токена бота и получение тестового токена для оплаты через Юкассу

• Создание бота через BotFather.

• Получение платежного токена Юкассы (рассмотрим, как получить тестовый и «боевой» токен).

2. Написание кода бота

Этот этап разделим на несколько подзадач:

• Описание таблиц базы данных (создание моделей).

• Миграции базы данных (преобразование моделей в реальные таблицы SQLite).

• Написание методов для взаимодействия с базой данных (добавление, удаление, изменение и получение данных).

• Реализация пользовательской логики (каталог, профиль пользователя, оплата, информация «о нас»).

• Реализация административной логики (статистика, добавление и удаление товаров).

3. Деплой бота

• Написание кода — это не все. Чтобы бот функционировал 24/7, мы разместим его на удаленном сервере.

• Для этого мы воспользуемся сервисом Amvera Cloud — удобным отечественным аналогом Heroku.

Шаги деплоя:

1. Подготовка файла с настройками (код будет предоставлен в статье).

2. Создание проекта в Amvera.

3. Загрузка файлов бота и настроек (вручную на сайте или через GIT).

4. Ожидание 3–4 минут для сборки и запуска бота.

Процесс займет не более 10 минут и будет понятен даже новичкам.

Настраиваем бота и подключаем оплату

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

Для выполнения этого шага мы воспользуемся ботом BotFather и платежной системой Юкасса. В результате вы получите полноценного Telegram-бота с интеграцией тестовой оплаты, а впоследствии сможете подключить реальные платежи.

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

Начнем с получения токена бота:

1. Откройте BotFather в Telegram.

2. Введите команду /newbot.

3. Укажите имя бота (можно на русском).

4. Придумайте уникальный логин для бота на латинице, оканчивающийся на BOT, bot или Bot.

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

Привязываем платежную систему к боту

Теперь перейдем к настройке приема тестовых платежей.

1. Зайдите в своего бота через BotFather.

2. Выберите раздел Payments.

3. Нажмите на Юкасса и выберите опцию Connect Test Юкасса.

4. BotFather перенаправит вас в бота Юкассы.

Внутри бота Юкассы:

• Если у вас уже есть аккаунт в Юкассе, выберите Войти и выдать доступ.

• Если аккаунта нет, нажмите Подключить Юкасса и следуйте инструкциям для регистрации.

Чтобы активировать тестовые платежи, убедитесь, что у вас есть тестовый магазин в личном кабинете Юкассы. Это обязательное условие для успешной привязки.

После успешного подключения вы увидите сообщение:

3. Получение тестового токена

Чтобы получить тестовый платежный токен:

1. Вернитесь в BotFather.

2. Откройте настройки своего бота и выберите Payments.

3. В разделе платежей вы найдете ваш тестовый токен.

После выполнения всех шагов у вас на руках должно быть два токена:

• Токен бота — для работы с Telegram API.

• Тестовый платежный токен — для интеграции с Юкассой.

Теперь мы готовы к следующим шагам разработки!

Организация проекта

Переходим к подготовке структуры и началу разработки Telegram-бота для магазина цифровых товаров. Начнем с общей настройки проекта и организации файловой структуры.

Шаг 1: Создаем проект и настраиваем окружение

1. В вашей любимой IDE создайте новый проект для бота.

2. Активируйте виртуальное окружение, чтобы изолировать зависимости.

После этого приступим к созданию базовой структуры проекта:

project   │   ├── bot/                # Основной код бота   │   ├── admin/          # Логика админ-панели   │   ├── user/           # Логика пользовательской части   │   ├── dao/            # Работа с базой данных   │   ├── config.py       # Настройки проекта   │   └── main.py         # Главный файл приложения   ├── data/               # Хранилище базы данных   ├── .env                # Переменные окружения (токены, настройки)   └── requirements.txt    # Зависимости проекта  

Шаг 2: Описание структуры папки bot

В папке bot сосредоточим весь основной код бота. Она включает:

• admin/ — директория для кода, связанного с админ-панелью бота.

• user/ — директория, где разместим логику пользовательской части бота, включая команды, меню и другие взаимодействия.

• dao/ — модуль для работы с базой данных. Здесь будут храниться модели и методы для взаимодействия с данными.

• config.py — файл настроек проекта, где укажем базовые параметры и пути.

• main.py — основной файл, с которого запускается приложение.

Шаг 3: Дополнительные файлы и директории

• data/ — папка для хранения базы данных.

• .env — файл для переменных окружения, таких как токен бота, токен платежной системы и другие конфиденциальные данные.

• requirements.txt — список всех зависимостей проекта для их быстрой установки.

Шаг 4: Будущие изменения в проекте

Позже, по мере работы над ботом, в проекте появятся:

• Миграции базы данных — их мы разместим в отдельной папке.

• Файл alembic.ini — для управления миграциями с помощью Alembic.

• Файлы в микросервисах (созданных пустых папках)

Настройка проекта и установка зависимостей

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

Установка зависимостей

В файл requirements.txt добавим следующие библиотеки:

aiogram==3.15.0 aiosqlite==0.20.0 loguru==0.7.2 pydantic-settings==2.7.0 SQLAlchemy==2.0.35 pydantic>=2.4.1,<2.10 alembic==1.14.0

Эти библиотеки обеспечивают основные функции бота, включая работу с Telegram API, базой данных, логирование и управление настройками. Установим их командой:

pip install -r requirements.txt

Настройка файла .env

Создайте файл .env в корне проекта и заполните его следующими переменными:

BOT_TOKEN=ВАШ_ТОКЕН_БОТА ADMIN_IDS=[AdminID1, AdminID2, AdminID3] PROVIDER_TOKEN=ТОКЕН_ПЛАТЕЖКИ

• BOT_TOKEN — токен вашего бота, который вы получили через BotFather.

• ADMIN_IDS — список Telegram ID администраторов, которые будут иметь доступ к админ-панели. Для получения ID можно использовать бота IDBot Finder Pro.

• PROVIDER_TOKEN — токен платежной системы, который мы подключили ранее.

Файл настроек: bot/config.py

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

Код файла config.py:

import os from typing import List from loguru import logger from aiogram import Bot, Dispatcher from aiogram.enums import ParseMode from aiogram.fsm.storage.memory import MemoryStorage from aiogram.client.default import DefaultBotProperties from pydantic_settings import BaseSettings, SettingsConfigDict   class Settings(BaseSettings):     BOT_TOKEN: str     ADMIN_IDS: List[int]     PROVIDER_TOKEN: str     FORMAT_LOG: str = "{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}"     LOG_ROTATION: str = "10 MB"     DB_URL: str = 'sqlite+aiosqlite:///data/db.sqlite3'     model_config = SettingsConfigDict(         env_file=os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", ".env")     ) 

Получаем параметры для загрузки переменных среды

settings = Settings()

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

bot = Bot(token=settings.BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML)) dp = Dispatcher(storage=MemoryStorage()) admins = settings.ADMIN_IDS  log_file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "log.txt") logger.add(log_file_path, format=settings.FORMAT_LOG, level="INFO", rotation=settings.LOG_ROTATION) database_url = settings.DB_URL 

Разбор кода

1. Импорты

   • Используются библиотеки для работы с переменными окружения (BaseSettings), логированием (loguru), Telegram API (aiogram) и базой данных.

2. Класс Settings

   • Наследуется от BaseSettings из pydantic-settings для автоматической загрузки переменных окружения из файла .env.

   • Ключевые параметры:

     • BOT_TOKEN, ADMIN_IDS, PROVIDER_TOKEN — обязательные переменные.

     • FORMAT_LOG и LOG_ROTATION — настройки логирования.

     • DB_URL — URL подключения к базе данных SQLite через aiosqlite.

   • model_config указывает путь к .env файлу.

3. Инициализация настроек

   • Создается объект settings, который загружает переменные из .env и предоставляет доступ к ним через атрибуты класса.

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

   • Bot и Dispatcher инициализируются на основе токена из settings.BOT_TOKEN.

   • Используется MemoryStorage для хранения FSM-состояний в памяти (в боевых проектах лучше использовать RedisStorage – подробно описывал в этой статье).

   • admins содержит список ID администраторов из переменной ADMIN_IDS.

5. Логирование

   • С помощью loguru создается лог-файл log.txt, в который записываются события с ротацией при достижении 10 МБ.

6. URL базы данных

   • Переменная database_url содержит путь для подключения к базе данных SQLite.

Теперь проект готов к следующему этапу разработки — реализации базовой логики.

Логика взаимодействия с базой данных

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

Полезные ресурсы для работы с SQLAlchemy 2 и Alembic

Перед тем как углубляться в код, рекомендую ознакомиться с моими статьями, которые помогут вам лучше понять работу с SQLAlchemy 2 и Alembic:

1. Асинхронный SQLAlchemy 2: простой пошаговый гайд по настройке, моделям, связям и миграциям с использованием Alembic.

2. Асинхронный SQLAlchemy 2: пошаговый гайд по управлению сессиями, добавлению и извлечению данных с Pydantic.

3. Асинхронный SQLAlchemy 2: улучшение кода, методы обновления и удаления данных.

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

Структура папки bot/dao

Весь код, связанный с базой данных, будет находиться в папке bot/dao. Создайте ее со следующей структурой:

├── dao/                 │   ├── __init__.py                # Пакетный файл для удобства импортов │   ├── database.py                # Настройки SQLAlchemy │   ├── models.py                  # Модели базы данных │   ├── base.py                    # Универсальный класс для взаимодействия с БД │   ├── dao.py                     # Специализированные DAO-классы │   └── database_middleware.py     # Мидлвари для управления сессиями базы данных 

Файл database.py

Файл database.py отвечает за настройки SQLAlchemy и создание базового класса для всех моделей.

from datetime import datetime from bot.config import database_url from sqlalchemy import func, TIMESTAMP, Integer from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase from sqlalchemy.ext.asyncio import AsyncAttrs, async_sessionmaker, create_async_engine, AsyncSession  # Создание асинхронного движка для подключения к БД engine = create_async_engine(url=database_url)  # Создание фабрики сессий async_session_maker = async_sessionmaker(engine, class_=AsyncSession)  # Базовый класс для моделей class Base(AsyncAttrs, DeclarativeBase):     __abstract__ = True  # Этот класс не будет создавать отдельную таблицу      # Общее поле "id" для всех таблиц     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)      # Поля времени создания и обновления записи     created_at: Mapped[datetime] = mapped_column(         TIMESTAMP, server_default=func.now()     )     updated_at: Mapped[datetime] = mapped_column(         TIMESTAMP, server_default=func.now(), onupdate=func.now()     )      # Автоматическое определение имени таблицы     @classmethod     @property     def __tablename__(cls) -> str:         return cls.__name__.lower() + 's' 

Краткий разбор файла:

• engine: Асинхронный движок для работы с базой данных. Подключается с помощью create_async_engine.

• async_session_maker: Фабрика для создания асинхронных сессий.

• Класс Base:

  • Базовый абстрактный класс для всех моделей.

  • Поле id: Общий первичный ключ.

  • Поля created_at и updated_at: Автоматическое управление временем создания и обновления записей.

Файл models.py

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

from typing import List from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy import BigInteger, Text, ForeignKey from bot.dao.database import Base   class User(Base):     telegram_id: Mapped[int] = mapped_column(BigInteger, unique=True, nullable=False)     username: Mapped[str | None]     first_name: Mapped[str | None]     last_name: Mapped[str | None]     purchases: Mapped[List['Purchase']] = relationship(         "Purchase",         back_populates="user",         cascade="all, delete-orphan"     )      def __repr__(self):         return f"<User(id={self.id}, telegram_id={self.telegram_id}, username='{self.username}')>"   class Category(Base):     __tablename__ = 'categories'      category_name: Mapped[str] = mapped_column(Text, nullable=False)     products: Mapped[List["Product"]] = relationship(         "Product",         back_populates="category",         cascade="all, delete-orphan"     )      def __repr__(self):         return f"<Category(id={self.id}, name='{self.category_name}')>"   class Product(Base):     name: Mapped[str] = mapped_column(Text)     description: Mapped[str] = mapped_column(Text)     price: Mapped[int]     file_id: Mapped[str | None] = mapped_column(Text)     category_id: Mapped[int] = mapped_column(ForeignKey('categories.id'))     hidden_content: Mapped[str] = mapped_column(Text)     category: Mapped["Category"] = relationship("Category", back_populates="products")     purchases: Mapped[List['Purchase']] = relationship(         "Purchase",         back_populates="product",         cascade="all, delete-orphan"     )      def __repr__(self):         return f"<Product(id={self.id}, name='{self.name}', price={self.price})>"   class Purchase(Base):     user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))     product_id: Mapped[int] = mapped_column(ForeignKey('products.id'))     price: Mapped[int]     payment_id: Mapped[str] = mapped_column(unique=True)     user: Mapped["User"] = relationship("User", back_populates="purchases")     product: Mapped["Product"] = relationship("Product", back_populates="purchases")      def __repr__(self):         return f"<Purchase(id={self.id}, user_id={self.user_id}, product_id={self.product_id}, date={self.created_at})>"

Основные моменты:

• User: Описывает пользователя Telegram. Содержит связи с покупками через relationship.

• Category: Категория товаров. Содержит связь с товарами.

• Product: Описывает товар с полями: название, описание, цена, файл и скрытый контент.

• Purchase: Информация о покупке с указанием пользователя, товара и стоимости.

Теперь мы готовы к тому, чтоб трансформировать модели базы данных в полноценную базу данных SQLite с таблицами в которых будут необходимые нам внутренние связи. Для этого мы воспользуемся интсрументом Alembic.

Настройка Alembic и создание первой миграции

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

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

Для начала переходим в директорию bot:

cd bot

Инициализируем Alembic с асинхронной поддержкой базы данных:

alembic init -t async migration

После выполнения команды появится папка migration и файл alembic.ini. Переместите alembic.ini в корневую директорию проекта для удобства работы.

Настройка файла alembic.ini

Откройте файл alembic.ini и измените строку:

script_location = migration

на:

script_location = bot/migration

Это упрощает использование миграций и запуск проекта из корневой директории.

Изменение env.py для подключения к базе данных

Теперь нам нужно внести изменения в файл bot/migration/env.py, чтобы Alembic мог корректно работать с нашей базой данных. Откройте файл и замените его содержимое следующим образом:

Было

import asyncio from logging.config import fileConfig from sqlalchemy import pool from sqlalchemy.engine import Connection from sqlalchemy.ext.asyncio import async_engine_from_config from alembic import context   config = context.config   if config.config_file_name is not None:     fileConfig(config.config_file_name)   target_metadata = None 

Стало

import sys from os.path import dirname, abspath  sys.path.insert(0, dirname(dirname(abspath(__file__))))  import asyncio from logging.config import fileConfig from sqlalchemy.ext.asyncio import async_engine_from_config from alembic import context from bot.dao.database import Base, database_url from bot.dao.models import Product, Purchase, User, Category  config = context.config config.set_main_option("sqlalchemy.url", database_url)  if config.config_file_name is not None:     fileConfig(config.config_file_name)  target_metadata = Base.metadata 

Оставшуюся часть файла можно оставить без изменений. В будущем автоматически будут отслеживаться добавленные модели таблиц.

Ключевые изменения:

1. Добавление корневой директории проекта в системный путь для корректного импорта модулей.

2. Указание URL подключения к базе данных.

3. Настройка метаданных для автоматической генерации миграций.

Создание первой миграции

Перейдите в корневую директорию проекта:

cd ../

Сгенерируйте файл миграции:

alembic revision --autogenerate -m "Initial revision"

Примените миграции для создания таблиц в базе данных:

alembic upgrade head

После выполнения этой команды в корне проекта появится файл db.sqlite3, содержащий таблицы users, purchases, products, и categories.

Работа с категориями

Для заполнения таблицы категорий можно вручную добавить данные. У каждой записи автоматически заполняются колонки id, created_at и updated_at. Вам нужно указать только category_name.

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

После «ручного» добавления категорий у меня получился следующий результат:

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

Вот как будет выглядеть реализация страницы каталога в нашем боте после интеграции в него таблицы категорий.

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

Пишем логику для универсальной работы с базой данных (класс BaseDao)

Для упрощения взаимодействия с базой данных и сокращения дублирования кода в проекте используется универсальный класс BaseDAO. Этот класс предоставляет базовые методы, которые можно переопределять и дополнять в дочерних классах.

Класс BaseDAO основан на шаблонном программировании и позволяет работать с любой моделью, унаследованной от Base.

Сам класс в исходном коде проекта больше, чем будет рассмотрен далее. Поэтому, если хотите получить полный код этого класса, то приглашаю в свой бесплатный телеграмм-канал «Легкий путь в Python». Там вы найдете не только полный исходный код сегодняшнего проекта, но и материал, который я не публикую на Хабре.

А пока опишем только те методы, которые будем использовать в нашем проекте:

Получится такой код:

from typing import List, Any, TypeVar, Generic from pydantic import BaseModel from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.future import select from sqlalchemy import update as sqlalchemy_update, delete as sqlalchemy_delete, func from loguru import logger from sqlalchemy.ext.asyncio import AsyncSession  from bot.dao.database import Base  # Объявляем типовой параметр T с ограничением, что это наследник Base T = TypeVar("T", bound=Base)   class BaseDAO(Generic[T]):     model: type[T]      @classmethod     async def find_one_or_none_by_id(cls, data_id: int, session: AsyncSession):         # Найти запись по ID         logger.info(f"Поиск {cls.model.__name__} с ID: {data_id}")         try:             query = select(cls.model).filter_by(id=data_id)             result = await session.execute(query)             record = result.scalar_one_or_none()             if record:                 logger.info(f"Запись с ID {data_id} найдена.")             else:                 logger.info(f"Запись с ID {data_id} не найдена.")             return record         except SQLAlchemyError as e:             logger.error(f"Ошибка при поиске записи с ID {data_id}: {e}")             raise      @classmethod     async def find_one_or_none(cls, session: AsyncSession, filters: BaseModel):         # Найти одну запись по фильтрам         filter_dict = filters.model_dump(exclude_unset=True)         logger.info(f"Поиск одной записи {cls.model.__name__} по фильтрам: {filter_dict}")         try:             query = select(cls.model).filter_by(**filter_dict)             result = await session.execute(query)             record = result.scalar_one_or_none()             if record:                 logger.info(f"Запись найдена по фильтрам: {filter_dict}")             else:                 logger.info(f"Запись не найдена по фильтрам: {filter_dict}")             return record         except SQLAlchemyError as e:             logger.error(f"Ошибка при поиске записи по фильтрам {filter_dict}: {e}")             raise      @classmethod     async def find_all(cls, session: AsyncSession, filters: BaseModel | None = None):         # Найти все записи по фильтрам         filter_dict = filters.model_dump(exclude_unset=True) if filters else {}         logger.info(f"Поиск всех записей {cls.model.__name__} по фильтрам: {filter_dict}")         try:             query = select(cls.model).filter_by(**filter_dict)             result = await session.execute(query)             records = result.scalars().all()             logger.info(f"Найдено {len(records)} записей.")             return records         except SQLAlchemyError as e:             logger.error(f"Ошибка при поиске всех записей по фильтрам {filter_dict}: {e}")             raise      @classmethod     async def add(cls, session: AsyncSession, values: BaseModel):         # Добавить одну запись         values_dict = values.model_dump(exclude_unset=True)         logger.info(f"Добавление записи {cls.model.__name__} с параметрами: {values_dict}")         new_instance = cls.model(**values_dict)         session.add(new_instance)         try:             await session.flush()             logger.info(f"Запись {cls.model.__name__} успешно добавлена.")         except SQLAlchemyError as e:             await session.rollback()             logger.error(f"Ошибка при добавлении записи: {e}")             raise e         return new_instance      @classmethod     async def delete(cls, session: AsyncSession, filters: BaseModel):         # Удалить записи по фильтру         filter_dict = filters.model_dump(exclude_unset=True)         logger.info(f"Удаление записей {cls.model.__name__} по фильтру: {filter_dict}")         if not filter_dict:             logger.error("Нужен хотя бы один фильтр для удаления.")             raise ValueError("Нужен хотя бы один фильтр для удаления.")          query = sqlalchemy_delete(cls.model).filter_by(**filter_dict)         try:             result = await session.execute(query)             await session.flush()             logger.info(f"Удалено {result.rowcount} записей.")             return result.rowcount         except SQLAlchemyError as e:             await session.rollback()             logger.error(f"Ошибка при удалении записей: {e}")             raise e      @classmethod     async def count(cls, session: AsyncSession, filters: BaseModel | None = None):         # Подсчитать количество записей         filter_dict = filters.model_dump(exclude_unset=True) if filters else {}         logger.info(f"Подсчет количества записей {cls.model.__name__} по фильтру: {filter_dict}")         try:             query = select(func.count(cls.model.id)).filter_by(**filter_dict)             result = await session.execute(query)             count = result.scalar()             logger.info(f"Найдено {count} записей.")             return count         except SQLAlchemyError as e:             logger.error(f"Ошибка при подсчете записей: {e}")             raise 

В своих статьях по SQLAlchemy я более подробно и детально рассматривал принципы взаимодействия с классом BaseDao.

Использование BaseDAO

Для каждой модели создается дочерний класс, унаследованный от BaseDAO. Например:

class UserDAO(BaseDAO[User]):     model = User 

Это позволяет вызывать методы напрямую, например:

user_info = await UserDAO.find_one_or_none(session=session, filters=filters) 

Если базовых методов недостаточно, в дочернем классе можно добавлять собственные методы.

Важный момент

В методах BaseDAO намеренно отсутствуют фиксации (commit) изменений в базе. Это позволяет выполнять несколько операций в рамках одной сессии и фиксировать их одним коммитом, если это необходимо. Такой подход особенно удобен в асинхронных проектах, например, в Telegram-ботах.

Как это работает в Telegram-боте?

1. Пользователь инициирует действие через бота.

2. Мидлварь создает и открывает сессию.

3. В функции обработки выполняются необходимые операции с базой данных.

4. После завершения операции сессия автоматически закрывается и, при необходимости, выполняет коммит.

Этот подход будет подробно рассмотрен на практике далее.

Дочерние классы DAO: работа с конкретными моделями

Для управления данными в проекте используются дочерние классы BaseDAO. Каждый из них привязан к определенной модели и может содержать дополнительные методы, если это требуется. В этом разделе мы разберем реализацию таких классов в файле bot/dao/dao.py.

Импорты

Сначала подключим необходимые модули и библиотеки:

from datetime import datetime, UTC, timedelta from typing import Optional, List, Dict  from loguru import logger from sqlalchemy import select, func, case from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload  from bot.dao.base import BaseDAO from bot.dao.models import User, Purchase, Category, Product 

Простые дочерние классы

Для некоторых моделей достаточно просто унаследоваться от BaseDAO и указать соответствующую модель. Это позволяет использовать все базовые методы BaseDAO, такие как добавление, удаление, поиск и подсчет записей.

class CategoryDao(BaseDAO[Category]):     model = Category   class ProductDao(BaseDAO[Product]):     model = Product 

Эти классы уже можно использовать, не добавляя дополнительных методов.

Работа с покупками: PurchaseDao

Модель покупок требует реализации метода для подсчета общей суммы всех покупок. Это полезно, например, для сбора статистики в админке.

class PurchaseDao(BaseDAO[Purchase]):     model = Purchase      @classmethod     async def get_full_summ(cls, session: AsyncSession) -> int:         """Получить общую сумму покупок."""         query = select(func.sum(cls.model.price).label('total_price'))         result = await session.execute(query)         total_price = result.scalars().one_or_none()         return total_price if total_price is not None else 0 

Этот метод возвращает общую сумму цен всех покупок. Если в базе отсутствуют покупки, метод возвращает 0.

Работа с пользователями: UserDAO

Модель пользователей требует реализации нескольких дополнительных методов для получения статистики и связанных данных. Рассмотрим их детально.

Статистика покупок пользователя

Этот метод возвращает общее количество покупок и их сумму для конкретного пользователя по его Telegram ID.

class UserDAO(BaseDAO[User]):     model = User      @classmethod     async def get_purchase_statistics(cls, session: AsyncSession, telegram_id: int) -> Optional[Dict[str, int]]:         try:             # Запрос для получения общего числа покупок и общей суммы             result = await session.execute(                 select(                     func.count(Purchase.id).label('total_purchases'),                     func.sum(Purchase.price).label('total_amount')                 ).join(User).filter(User.telegram_id == telegram_id)             )             stats = result.one_or_none()              if stats is None:                 return None              total_purchases, total_amount = stats             return {                 'total_purchases': total_purchases,                 'total_amount': total_amount or 0  # Обработка случая, когда сумма может быть None             }          except SQLAlchemyError as e:             # Обработка ошибок при работе с базой данных             print(f"Ошибка при получении статистики покупок пользователя: {e}")             return None 

Список покупок пользователя

Метод возвращает список всех покупок пользователя с детализацией по продуктам.

   @classmethod     async def get_purchased_products(cls, session: AsyncSession, telegram_id: int) -> Optional[List[Purchase]]:         try:             # Запрос для получения пользователя с его покупками и связанными продуктами             result = await session.execute(                 select(User)                 .options(                     selectinload(User.purchases).selectinload(Purchase.product)                 )                 .filter(User.telegram_id == telegram_id)             )             user = result.scalar_one_or_none()              if user is None:                 return None              return user.purchases          except SQLAlchemyError as e:             # Обработка ошибок при работе с базой данных             print(f"Ошибка при получении информации о покупках пользователя: {e}")             return None

Этот метод загружает связанные покупки и продукты через ORM-загрузку (selectinload), чтобы минимизировать количество запросов к базе.

Общая статистика пользователей

Этот метод собирает данные о количестве пользователей, зарегистрированных за различные временные периоды.

   @classmethod     async def get_statistics(cls, session: AsyncSession):         try:             now = datetime.now(UTC)              query = select(                 func.count().label('total_users'),                 func.sum(case((cls.model.created_at >= now - timedelta(days=1), 1), else_=0)).label('new_today'),                 func.sum(case((cls.model.created_at >= now - timedelta(days=7), 1), else_=0)).label('new_week'),                 func.sum(case((cls.model.created_at >= now - timedelta(days=30), 1), else_=0)).label('new_month')             )              result = await session.execute(query)             stats = result.fetchone()              statistics = {                 'total_users': stats.total_users,                 'new_today': stats.new_today,                 'new_week': stats.new_week,                 'new_month': stats.new_month             }              logger.info(f"Статистика успешно получена: {statistics}")             return statistics         except SQLAlchemyError as e:             logger.error(f"Ошибка при получении статистики: {e}")             raise

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

Теперь нам остается последний этап в настройке взаимодействия с базой данных через SQLAlchemy, а именно, описание мидлвари.

Создаем мидлвари для управления сессиями базы данных

Что такое мидлвари?

Мидлвари в Telegram-ботах — это промежуточные слои логики, которые выполняются между получением события (например, сообщения или callback-запроса) и обработкой его хендлером. Они позволяют изменять данные события, добавлять дополнительные параметры или выполнять сторонние действия (например, создание сессии для работы с базой данных).

Почему нам нужны мидлвари для управления сессиями?

В нашем проекте мы реализуем автоматическое управление сессиями базы данных. Это означает, что:

• Сессия автоматически открывается перед обработкой события.

• В зависимости от необходимости изменения данных, сессия либо фиксируется (коммитится), либо откатывается (rollback).

• После завершения обработки сессия автоматически закрывается.

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

Работаем с файлом bot/dao/database_middleware.py

Начнем с необходимых импортов:

from typing import Callable, Dict, Any, Awaitable from aiogram import BaseMiddleware from aiogram.types import Message, CallbackQuery from bot.dao.database import async_session_maker 

Реализация базового класса

Основу логики управления сессиями мы поместим в базовый класс BaseDatabaseMiddleware. Он будет:

1. Открывать сессию перед обработкой события.

2. Передавать сессию в data — специальный словарь, используемый в Aiogram для передачи данных между мидлварями и хендлерами.

3. Закрывать сессию автоматически, вне зависимости от исхода обработки.

Вот как выглядит этот класс:

class BaseDatabaseMiddleware(BaseMiddleware):     async def __call__(         self,         handler: Callable[[Message | CallbackQuery, Dict[str, Any]], Awaitable[Any]],         event: Message | CallbackQuery,         data: Dict[str, Any]     ) -> Any:         async with async_session_maker() as session:             self.set_session(data, session)  # Устанавливаем сессию             try:                 result = await handler(event, data)  # Обрабатываем событие                 await self.after_handler(session)  # Дополнительные действия (например, коммит)                 return result             except Exception as e:                 await session.rollback()  # Откат изменений в случае ошибки                 raise e             finally:                 await session.close()  # Закрываем сессию      def set_session(self, data: Dict[str, Any], session) -> None:         """Метод для установки сессии в данные. Реализуется в дочерних классах."""         raise NotImplementedError("Этот метод должен быть реализован в подклассах.")      async def after_handler(self, session) -> None:         """Метод для выполнения действий после обработки события. По умолчанию ничего не делает."""         pass 

Дочерние классы для управления сессиями

Теперь создадим два дочерних класса с конкретной реализацией логики:

1. Мидлварь для сессии без коммита

Эта мидлварь просто передает сессию в data без выполнения коммита.

class DatabaseMiddlewareWithoutCommit(BaseDatabaseMiddleware):     def set_session(self, data: Dict[str, Any], session) -> None:         """Устанавливаем сессию без коммита."""         data['session_without_commit'] = session 

2. Мидлварь для сессии с коммитом

Эта мидлварь дополнительно фиксирует изменения в базе данных после успешной обработки события.

class DatabaseMiddlewareWithCommit(BaseDatabaseMiddleware):     def set_session(self, data: Dict[str, Any], session) -> None:         """Устанавливаем сессию с коммитом."""         data['session_with_commit'] = session      async def after_handler(self, session) -> None:         """Фиксируем изменения после обработки события."""         await session.commit() 

Как это работает?

Каждая из мидлварей автоматически:

1. Открывает сессию через async_session_maker().

2. Добавляет сессию в данные для последующего использования в хендлерах.

3. При завершении обработки:

   • Если используется DatabaseMiddlewareWithCommit, выполняется коммит.

   • Если используется DatabaseMiddlewareWithoutCommit, изменения в базе остаются необработанными.

4. В случае ошибки откатывает изменения и закрывает сессию.

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

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

Пишем админ-панель бота

Начнем с описания админ-панели. Это логично, так как админка будет содержать функционал для добавления товаров в наш магазин. Без товаров магазина не существует, поэтому реализуем эту часть в первую очередь.

Структура файлов админ-панели

Мы будем работать с папкой bot/admin, в которой создадим следующую структуру:

├── admin/                 │   ├── __init__.py           # Пакетный файл для удобства импортов │   ├── admin.py              # Основной файл с методами админ-панели │   ├── kbs.py                # Описание клавиатур админ-панели │   ├── schemas.py            # Pydantic-схемы для работы с данными │   └── utils.py              # Вспомогательные утилиты для админ-панели 

Эта структура компактная и удобная для небольших проектов. В крупных проектах вы можете разбивать логику на дополнительные папки. Например, я часто помещаю в такие модули файлы dao.py и models.py, где описываю конкретные модели и дочерние классы DAO, чтобы модуль можно было масштабировать. Подобный подход также используется при создании микросервисов на FastAPI.

Файл utils.py

Начнем с утилит. Здесь мы реализуем метод для безопасного удаления предыдущих сообщений. Это поможет избежать засорения диалога администратора с ботом. Метод будет выглядеть так:

from aiogram.fsm.context import FSMContext from aiogram.types import Message from loguru import logger  from bot.config import bot   async def process_dell_text_msg(message: Message, state: FSMContext):     data = await state.get_data()     last_msg_id = data.get('last_msg_id')      try:         if last_msg_id:             await bot.delete_message(chat_id=message.from_user.id, message_id=last_msg_id)         else:             logger.warning("Ошибка: Не удалось найти идентификатор последнего сообщения для удаления.")         await message.delete()      except Exception as e:         logger.error(f"Произошла ошибка при удалении сообщения: {str(e)}") 

Объяснение метода:

• Мы передаем в метод объект message и текущее состояние state.

• Состояние используется для получения ID последнего сообщения (last_msg_id), которое нужно удалить.

• Текущее сообщение удаляется через метод await message.delete().

• Если что-то пошло не так, логируем ошибку.

Файл schemas.py

Здесь опишем Pydantic-схемы, которые используются для валидации и добавления данных в таблицы, связанные с админкой. Напоминаю, что в BaseDAO в качестве аргументов для добавления или фильтрации данных мы используем экземпляры BaseModel Pydantic, а не словари.

from pydantic import BaseModel, Field   class ProductIDModel(BaseModel):     id: int   class ProductModel(BaseModel):     name: str = Field(..., min_length=5)     description: str = Field(..., min_length=5)     price: int = Field(..., gt=0)     category_id: int = Field(..., gt=0)     file_id: str | None = None     hidden_content: str = Field(..., min_length=5) 

Объяснение:

• ProductIDModel используется для передачи ID продукта.

• ProductModel описывает структуру данных для продукта, включая название, описание, цену, ID категории, файл и скрытый контент.

• Параметры Field задают дополнительные ограничения, например, минимальную длину строки или положительное значение цены.

Файл kbs.py

Теперь создадим файл с клавиатурами. Все кнопки будут реализованы с помощью InlineKeyboardBuilder, который предоставляет удобный способ работы с инлайн-кнопками в Aiogram 3.

from typing import List from aiogram.types import InlineKeyboardMarkup from aiogram.utils.keyboard import InlineKeyboardBuilder from bot.dao.models import Category   def catalog_admin_kb(catalog_data: List[Category]) -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     for category in catalog_data:         kb.button(text=category.category_name, callback_data=f"add_category_{category.id}")     kb.button(text="Отмена", callback_data="admin_panel")     kb.adjust(2)     return kb.as_markup()   def admin_send_file_kb() -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     kb.button(text="Без файла", callback_data="without_file")     kb.button(text="Отмена", callback_data="admin_panel")     kb.adjust(2)     return kb.as_markup()   def admin_kb() -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     kb.button(text="📊 Статистика", callback_data="statistic")     kb.button(text="🛍️ Управлять товарами", callback_data="process_products")     kb.button(text="🏠 На главную", callback_data="home")     kb.adjust(2)     return kb.as_markup()   def admin_kb_back() -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     kb.button(text="⚙️ Админ панель", callback_data="admin_panel")     kb.button(text="🏠 На главную", callback_data="home")     kb.adjust(1)     return kb.as_markup()   def dell_product_kb(product_id: int) -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     kb.button(text="🗑️ Удалить", callback_data=f"dell_{product_id}")     kb.button(text="⚙️ Админ панель", callback_data="admin_panel")     kb.button(text="🏠 На главную", callback_data="home")     kb.adjust(2, 2, 1)     return kb.as_markup()   def product_management_kb() -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     kb.button(text="➕ Добавить товар", callback_data="add_product")     kb.button(text="🗑️ Удалить товар", callback_data="delete_product")     kb.button(text="⚙️ Админ панель", callback_data="admin_panel")     kb.button(text="🏠 На главную", callback_data="home")     kb.adjust(2, 2, 1)     return kb.as_markup()   def cancel_kb_inline() -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     kb.button(text="Отмена", callback_data="cancel")     return kb.as_markup()   def admin_confirm_kb() -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     kb.button(text="Все верно", callback_data="confirm_add")     kb.button(text="Отмена", callback_data="admin_panel")     kb.adjust(1)     return kb.as_markup() 

Объяснение:

• Все клавиатуры оформлены в виде функций, возвращающих объект InlineKeyboardMarkup.

• Используем современные методы из Aiogram 3 для динамического создания кнопок.

• Кнопки группируются по 2-3 в ряду для удобства отображения.

Теперь можно приступать к написанию основного кода админки.

Описываем логику админ-панели

Прежде чем приступить к описанию кода, я хочу предположить, что вы уже знакомы с фреймворком Aiogram 3. Если это не так, то рекомендую ознакомиться с моими статьями, в которых я подробно, от начала до конца, рассказываю о создании Telegram-ботов на основе Aiogram 3. Вы можете найти все статьи в хронологическом порядке по следующей ссылке: Полный список статей.

Далее, я буду исходить из того, что у вас есть базовые знания взаимодействия с Aiogram 3.

Теперь мы готовы к описанию основной логики админ-панели. Код будем писать в файле bot/admin/admin.py

Начнем с импортов и инициализации роутера:

import asyncio from aiogram import Router, F from aiogram.fsm.context import FSMContext from aiogram.fsm.state import StatesGroup, State from aiogram.types import CallbackQuery, Message from sqlalchemy.ext.asyncio import AsyncSession from bot.config import settings, bot from bot.dao.dao import UserDAO, ProductDao, CategoryDao, PurchaseDao from bot.admin.kbs import admin_kb, admin_kb_back, product_management_kb, cancel_kb_inline, catalog_admin_kb, \     admin_send_file_kb, admin_confirm_kb, dell_product_kb from bot.admin.schemas import ProductModel, ProductIDModel from bot.admin.utils import process_dell_text_msg  admin_router = Router() 

Теперь опишем класс в котором будем хранить наши состояния FSM:

class AddProduct(StatesGroup):     name = State()     description = State()     price = State()     file_id = State()     category_id = State()     hidden_content = State()     confirm_add = State() 

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

Опишем первую функцию, которая будет выполняться при входе в админ панель:

@admin_router.callback_query(F.data == "admin_panel", F.from_user.id.in_(settings.ADMIN_IDS)) async def start_admin(call: CallbackQuery):     await call.answer('Доступ в админ-панель разрешен!')     await call.message.edit_text(         text="Вам разрешен доступ в админ-панель. Выберите необходимое действие.",         reply_markup=admin_kb()     ) 

Тут я использовал магические фильтры Aiogram 3 для проверки того, что данный метод вызвал администратор:

F.from_user.id.in_(settings.ADMIN_IDS))

И для проверки того, что была вызвана call_data – «admin_panel»:

F.data == "admin_panel"

Список телеграмм айди администраторов мы берем с переменной ADMIN_IDS.

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

Опишем метод, который будет возвращать статистику по боту.

@admin_router.callback_query(F.data == 'statistic', F.from_user.id.in_(settings.ADMIN_IDS)) async def admin_statistic(call: CallbackQuery, session_without_commit: AsyncSession):     await call.answer('Запрос на получение статистики...')     await call.answer('📊 Собираем статистику...')      stats = await UserDAO.get_statistics(session=session_without_commit)     total_summ = await PurchaseDao.get_full_summ(session=session_without_commit)     stats_message = (         "📈 Статистика пользователей:\n\n"         f"👥 Всего пользователей: {stats['total_users']}\n"         f"🆕 Новых за сегодня: {stats['new_today']}\n"         f"📅 Новых за неделю: {stats['new_week']}\n"         f"📆 Новых за месяц: {stats['new_month']}\n\n"         f"💰 Общая сумма заказов: {total_summ} руб.\n\n"         "🕒 Данные актуальны на текущий момент."     )     await call.message.edit_text(         text=stats_message,         reply_markup=admin_kb()     ) 

Тут из примечательного — это вызов сессии:

session_without_commit: AsyncSession

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

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

Теперь разберем главную страницу блока для управления товарами. Выглядит она так:

@admin_router.callback_query(F.data == 'process_products', F.from_user.id.in_(settings.ADMIN_IDS)) async def admin_process_products(call: CallbackQuery, session_without_commit: AsyncSession):     await call.answer('Режим управления товарами')     all_products_count = await ProductDao.count(session=session_without_commit)     await call.message.edit_text(         text=f"На данный момент в базе данных {all_products_count} товаров. Что будем делать?",         reply_markup=product_management_kb()     ) 

Тут из примечательного то, что для ProductDao мы вызвали метод count, который был унаследован из родительского класса BaseDao.

Я добавил только 2 сценария: добавление товара и удаление товара. В целом, для практического закрепления материала статьи вы можете самостоятельно описать метод для редактирования товаров.

Сразу разберем логику удаления товаров с базы данных, так как она короткая.

При нажатии на кнопку «Удалить товар» появляется список всех товаров нашего магазина. В этом списке есть встроенная кнопка «Удалить», в которой хранится уникальный идентификатор товара. В реальном проекте можно было бы добавить сортировку по категориям, внутренний поисковик и другие функции, но я решил не усложнять задачу.

Подгрузка товаров выглядит так:

@admin_router.callback_query(F.data == 'delete_product', F.from_user.id.in_(settings.ADMIN_IDS)) async def admin_process_start_dell(call: CallbackQuery, session_without_commit: AsyncSession):     await call.answer('Режим удаления товаров')     all_products = await ProductDao.find_all(session=session_without_commit)      await call.message.edit_text(         text=f"На данный момент в базе данных {len(all_products)} товаров. Для удаления нажмите на кнопку ниже"     )     for product_data in all_products:         file_id = product_data.file_id         file_text = "📦 Товар с файлом" if file_id else "📄 Товар без файла"          product_text = (f'🛒 Описание товара:\n\n'                         f'🔹 <b>Название товара:</b> <b>{product_data.name}</b>\n'                         f'🔹 <b>Описание:</b>\n\n<b>{product_data.description}</b>\n\n'                         f'🔹 <b>Цена:</b> <b>{product_data.price} ₽</b>\n'                         f'🔹 <b>Описание (закрытое):</b>\n\n<b>{product_data.hidden_content}</b>\n\n'                         f'<b>{file_text}</b>')         if file_id:             await call.message.answer_document(document=file_id, caption=product_text,                                                reply_markup=dell_product_kb(product_data.id))         else:             await call.message.answer(text=product_text, reply_markup=dell_product_kb(product_data.id))  

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

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

await call.message.answer_document(document=file_id) 

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

Далее, после клика на кнопку «Удалить» запускается следующая функция:

@admin_router.callback_query(F.data.startswith('dell_'), F.from_user.id.in_(settings.ADMIN_IDS)) async def admin_process_start_dell(call: CallbackQuery, session_with_commit: AsyncSession):     product_id = int(call.data.split('_')[-1])     await ProductDao.delete(session=session_with_commit, filters=ProductIDModel(id=product_id))     await call.answer(f"Товар с ID {product_id} удален!", show_alert=True)     await call.message.delete()  

Тут обратите внимание, что мы вызываем:

session_with_commit: AsyncSession 

Так как удаление информации с базы данных подразумевает фиксацию (commit).

В этом проекте не предусмотрен механизм ограничения. В реальных приложениях бот должен спрашивать: «Вы уверены, что хотите удалить?» — и только после подтверждения пользователя удалять данные. Я решил не тратить время на разработку такого функционала.

Описываем логику добавления товаров

В админ панели самая важная и одновременно самая сложная логика выстраивается вокруг добавления товара в базу данных.

Тут мы будем использовать технологию Finite State Machine (конечные автоматы) или, другими словами, сценарий «опроса» администратора при добавлении товара.

Мы уже подготовили класс для FSM и теперь останется только его интегрировать.

Хочу подчеркнуть, что в этой статье я не использовал популярную надстройку над aiogram — Aiogram Dialog. Вместо этого я реализовал всё через чистый FSM Aigram 3.

В конце статьи вас ждёт голосование. Если вы хотите, чтобы я подробно разобрал библиотеку Aiogram Dialog на практическом примере, примите участие в голосовании.

Хочу также обратить ваше внимание на ещё один важный момент. В настоящее время в качестве хранилища мы используем MemoryStorage. Имейте в виду, что в этом случае информация, полученная в ходе вашего опроса, будет храниться в памяти бота. Если бот перезагрузится, он «забудет» все, о чём вы говорили.

Конечно, в боевых проектах это не допустимо и нужно использовать более надежные хранилища, в частности RedisStorage.

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

@admin_router.callback_query(F.data == "cancel", F.from_user.id.in_(settings.ADMIN_IDS)) async def admin_process_cancel(call: CallbackQuery, state: FSMContext):     await state.clear()     await call.answer('Отмена сценария добавления товара')     await call.message.delete()     await call.message.answer(         text="Отмена добавления товара.",         reply_markup=admin_kb_back()     ) 

Данную функцию поместите где-то вначале файла admin.py.

Тут важно понимать, что теперь во всех функциях по умолчанию установлен state=[*]. Если вы писали на aiogram 2, то знаете, что раньше нужно было самостоятельно вызывать это универсальное состояние, теперь оно по умолчанию.

Кроме того, в Aiogram 3 теперь по умолчанию стоит обработчик на тип контента ANY. Поэтому не забывайте явно указывать какой тип контента должен быть обработан через F-фильтры например.

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

@admin_router.callback_query(F.data == 'add_product', F.from_user.id.in_(settings.ADMIN_IDS)) async def admin_process_add_product(call: CallbackQuery, state: FSMContext):     await call.answer('Запущен сценарий добавления товара.')     await call.message.delete()     msg = await call.message.answer(text="Для начала укажите имя товара: ", reply_markup=cancel_kb_inline())     await state.update_data(last_msg_id=msg.message_id)     await state.set_state(AddProduct.name)   @admin_router.message(F.text, F.from_user.id.in_(settings.ADMIN_IDS), AddProduct.name) async def admin_process_name(message: Message, state: FSMContext):     await state.update_data(name=message.text)     await process_dell_text_msg(message, state)     msg = await message.answer(text="Теперь дайте короткое описание товару: ", reply_markup=cancel_kb_inline())     await state.update_data(last_msg_id=msg.message_id)     await state.set_state(AddProduct.description)   @admin_router.message(F.text, F.from_user.id.in_(settings.ADMIN_IDS), AddProduct.description) async def admin_process_description(message: Message, state: FSMContext, session_without_commit: AsyncSession):     await state.update_data(description=message.html_text)     await process_dell_text_msg(message, state)     catalog_data = await CategoryDao.find_all(session=session_without_commit)     msg = await message.answer(text="Теперь выберите категорию товара: ", reply_markup=catalog_admin_kb(catalog_data))     await state.update_data(last_msg_id=msg.message_id)     await state.set_state(AddProduct.category_id)   @admin_router.callback_query(F.data.startswith("add_category_"),                              F.from_user.id.in_(settings.ADMIN_IDS),                              AddProduct.category_id) async def admin_process_category(call: CallbackQuery, state: FSMContext):     category_id = int(call.data.split("_")[-1])     await state.update_data(category_id=category_id)     await call.answer('Категория товара успешно выбрана.')     msg = await call.message.edit_text(text="Введите цену товара: ", reply_markup=cancel_kb_inline())     await state.update_data(last_msg_id=msg.message_id)     await state.set_state(AddProduct.price)   @admin_router.message(F.text, F.from_user.id.in_(settings.ADMIN_IDS), AddProduct.price) async def admin_process_price(message: Message, state: FSMContext):     try:         price = int(message.text)         await state.update_data(price=price)         await process_dell_text_msg(message, state)         msg = await message.answer(             text="Отправьте файл (документ), если требуется или нажмите на 'БЕЗ ФАЙЛА', если файл не требуется",             reply_markup=admin_send_file_kb()         )         await state.update_data(last_msg_id=msg.message_id)         await state.set_state(AddProduct.file_id)     except ValueError:         await message.answer(text="Ошибка! Необходимо ввести числовое значение для цены.")         return   @admin_router.callback_query(F.data == "without_file", F.from_user.id.in_(settings.ADMIN_IDS), AddProduct.file_id) async def admin_process_without_file(call: CallbackQuery, state: FSMContext):     await state.update_data(file_id=None)     await call.answer('Файл не выбран.')     msg = await call.message.edit_text(         text="Теперь отправьте контент, который отобразится после покупки товара внутри карточки",         reply_markup=cancel_kb_inline())     await state.update_data(last_msg_id=msg.message_id)     await state.set_state(AddProduct.hidden_content)   @admin_router.message(F.document, F.from_user.id.in_(settings.ADMIN_IDS), AddProduct.file_id) async def admin_process_without_file(message: Message, state: FSMContext):     await state.update_data(file_id=message.document.file_id)     await process_dell_text_msg(message, state)     msg = await message.answer(         text="Теперь отправьте контент, который отобразится после покупки товара внутри карточки",         reply_markup=cancel_kb_inline())     await state.update_data(last_msg_id=msg.message_id)     await state.set_state(AddProduct.hidden_content)   @admin_router.message(F.text, F.from_user.id.in_(settings.ADMIN_IDS), AddProduct.hidden_content) async def admin_process_hidden_content(message: Message, state: FSMContext, session_without_commit: AsyncSession):     await state.update_data(hidden_content=message.html_text)      product_data = await state.get_data()     category_info = await CategoryDao.find_one_or_none_by_id(session=session_without_commit,                                                              data_id=product_data.get("category_id"))      file_id = product_data.get("file_id")     file_text = "📦 Товар с файлом" if file_id else "📄 Товар без файла"      product_text = (f'🛒 Проверьте, все ли корректно:\n\n'                     f'🔹 <b>Название товара:</b> <b>{product_data["name"]}</b>\n'                     f'🔹 <b>Описание:</b>\n\n<b>{product_data["description"]}</b>\n\n'                     f'🔹 <b>Цена:</b> <b>{product_data["price"]} ₽</b>\n'                     f'🔹 <b>Описание (закрытое):</b>\n\n<b>{product_data["hidden_content"]}</b>\n\n'                     f'🔹 <b>Категория:</b> <b>{category_info.category_name} (ID: {category_info.id})</b>\n\n'                     f'<b>{file_text}</b>')     await process_dell_text_msg(message, state)      if file_id:         msg = await message.answer_document(document=file_id, caption=product_text, reply_markup=admin_confirm_kb())     else:         msg = await message.answer(text=product_text, reply_markup=admin_confirm_kb())     await state.update_data(last_msg_id=msg.message_id)     await state.set_state(AddProduct.confirm_add)   @admin_router.callback_query(F.data == "confirm_add", F.from_user.id.in_(settings.ADMIN_IDS)) async def admin_process_confirm_add(call: CallbackQuery, state: FSMContext, session_with_commit: AsyncSession):     await call.answer('Приступаю к сохранению файла!')     product_data = await state.get_data()     await bot.delete_message(chat_id=call.from_user.id, message_id=product_data["last_msg_id"])     del product_data["last_msg_id"]     await ProductDao.add(session=session_with_commit, values=ProductModel(**product_data))     await call.message.answer(text="Товар успешно добавлен в базу данных!", reply_markup=admin_kb()) 

Если будут вопросы — пишите тут в комментариях или в сообществе «Легкий путь в Python». Там у нас дружелюбная атмосфера и почти 2000 единомышленников.

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

На этом наша админ-панель готова, а это значит, что мы можем приступать к пользовательской части.

Пишем пользовательскую часть бота

Структура файлов пользовательской части

Мы будем работать с папкой bot/user, в которой создадим следующую структуру:

├── user/                 │   ├── __init__.py           # Пакетный файл для удобства импортов │   ├── user_router.py        # Файл в котором опишем общие методы для пользователя: профиль, просмотр купленных товаров, блок «О нас» │   ├──catalog_router.py      # Файл в котором опишем методы для взаимодействия с каталогом: просмотр товаров, страница каталога, покупка товаров │   ├── kbs.py                # Описание клавиатур пользователя │   └── schemas.py            # Pydantic-схемы для работы с данными 

Как видите, для пользователя мы реализуем два роутера.

Начнем со вспомогательных файлов.

Клавиатуры пользователя (kbs.py)

from typing import List from aiogram.types import InlineKeyboardMarkup, ReplyKeyboardMarkup, InlineKeyboardButton from aiogram.utils.keyboard import InlineKeyboardBuilder, ReplyKeyboardBuilder from bot.config import settings from bot.dao.models import Category   def main_user_kb(user_id: int) -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     kb.button(text="👤 Мои покупки", callback_data="my_profile")     kb.button(text="🛍 Каталог", callback_data="catalog")     kb.button(text="ℹ️ О магазине", callback_data="about")     kb.button(text="🌟 Поддержать автора 🌟", url='https://t.me/tribute/app?startapp=deLN')     if user_id in settings.ADMIN_IDS:         kb.button(text="⚙️ Админ панель", callback_data="admin_panel")     kb.adjust(1)     return kb.as_markup()   def catalog_kb(catalog_data: List[Category]) -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     for category in catalog_data:         kb.button(text=category.category_name, callback_data=f"category_{category.id}")     kb.button(text="🏠 На главную", callback_data="home")     kb.adjust(2)     return kb.as_markup()   def purchases_kb() -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     kb.button(text="🗑 Смотреть покупки", callback_data="purchases")     kb.button(text="🏠 На главную", callback_data="home")     kb.adjust(1)     return kb.as_markup()   def product_kb(product_id, price) -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     kb.button(text="💸 Купить", callback_data=f"buy_{product_id}_{price}")     kb.button(text="🛍 Назад", callback_data="catalog")     kb.button(text="🏠 На главную", callback_data="home")     kb.adjust(2)     return kb.as_markup()   def get_product_buy_kb(price) -> InlineKeyboardMarkup:     return InlineKeyboardMarkup(inline_keyboard=[         [InlineKeyboardButton(text=f'Оплатить {price}₽', pay=True)],         [InlineKeyboardButton(text='Отменить', callback_data='home')]     ])  

Из всех клавиатур тут выбивается только клавиатура с кнопкой «Оплатить»

def get_product_buy_kb(price) -> InlineKeyboardMarkup:     return InlineKeyboardMarkup(inline_keyboard=[         [InlineKeyboardButton(text=f'Оплатить {price}₽', pay=True)],         [InlineKeyboardButton(text='Отменить', callback_data='home')]     ]) 

Я намеренно оставил ее в таком виде, чтоб продемонстрировать вам, что классический подход описания клавиатур с использованием InlineKeyboardMarkup все ещё поддерживается в aiogram 3, хотя все чаще используется Builder.

Кроме того, в этой клавиатуре мы используем необычную и, возможно, незнакомую вам кнопку:

InlineKeyboardButton(text=f'Оплатить {price}₽', pay=True)

Тут важно понимать, что в одной клавиатуре может быть только одна платежная кнопка.

В остальном, если вы знакомы с базой по Aiogram 3, то общая логика этого кода вам должна быть понятна.

Схемы Pydantic (kbs.py)

 from pydantic import BaseModel, ConfigDict, Field   class TelegramIDModel(BaseModel):     telegram_id: int      model_config = ConfigDict(from_attributes=True)   class UserModel(TelegramIDModel):     username: str | None     first_name: str | None     last_name: str | None   class ProductIDModel(BaseModel):     id: int   class ProductCategoryIDModel(BaseModel):     category_id: int   class PaymentData(BaseModel):     user_id: int = Field(..., description="ID пользователя Telegram")     payment_id: str = Field(..., max_length=255, description="Уникальный ID платежа")     price: int = Field(..., description="Сумма платежа в рублях")     product_id: int = Field(..., description="ID товара") 

В этом наборе есть как фильтрующие схемы, такие как TelegramIDModel, ProductIDModel и ProductCategoryIDModel, так и схемы, которые помогут нам проверить данные перед их сохранением в таблицах. К ним относятся:

• PaymentData — для хранения информации о платежах;

• UserModel — для добавления пользователей в базу данных.

Теперь перейдём к описанию логики процесса покупки. Ведь без покупки не было бы смысла заходить в личный кабинет и в раздел покупок, не так ли?

Работем с роутером каталога для польователя (catalog_router.py)

Выполним импорты и назначим роутер:

from aiogram import Router, F from aiogram.enums import ContentType from aiogram.types import Message, CallbackQuery, LabeledPrice, PreCheckoutQuery from loguru import logger from sqlalchemy.ext.asyncio import AsyncSession from bot.config import bot, settings from bot.dao.dao import UserDAO, CategoryDao, ProductDao, PurchaseDao from bot.user.kbs import main_user_kb, catalog_kb, product_kb, get_product_buy_kb from bot.user.schemas import TelegramIDModel, ProductCategoryIDModel, PaymentData  catalog_router = Router() 

Теперь опишем первую страницу, которую будет видеть пользователь при входе в раздел каталога.

@catalog_router.callback_query(F.data == "catalog") async def page_catalog(call: CallbackQuery, session_without_commit: AsyncSession):     await call.answer("Загрузка каталога...")     catalog_data = await CategoryDao.find_all(session=session_without_commit)      await call.message.edit_text(         text="Выберите категорию товаров:",         reply_markup=catalog_kb(catalog_data)     ) 

Внимание заслуживает разве что формат формирование клавиатуры с каталогом:

def catalog_kb(catalog_data: List[Category]) -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     for category in catalog_data:         kb.button(text=category.category_name, callback_data=f"category_{category.id}")     kb.button(text="🏠 На главную", callback_data="home")     kb.adjust(2)     return kb.as_markup()  

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

Далее, через цикл, мы делим каждую категорию по 2 кнопки в ряд, добавляя в конце кнопку «🏠 На главную». В результате получаем такую страницу:

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

Теперь опишем страницу, которую увидит пользователь, кликнув на ту или иную категорию:

@catalog_router.callback_query(F.data.startswith("category_")) async def page_catalog_products(call: CallbackQuery, session_without_commit: AsyncSession):     category_id = int(call.data.split("_")[-1])     products_category = await ProductDao.find_all(session=session_without_commit,                                                   filters=ProductCategoryIDModel(category_id=category_id))     count_products = len(products_category)     if count_products:         await call.answer(f"В данной категории {count_products} товаров.")         for product in products_category:             product_text = (                 f"📦 <b>Название товара:</b> {product.name}\n\n"                 f"💰 <b>Цена:</b> {product.price} руб.\n\n"                 f"📝 <b>Описание:</b>\n<i>{product.description}</i>\n\n"                 f"━━━━━━━━━━━━━━━━━━"             )             await call.message.answer(                 product_text,                 reply_markup=product_kb(product.id, product.price)             )     else:         await call.answer("В данной категории нет товаров.")  

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

Обратите внимание, при отображении текста сохраняется форматирование, которое было задано на этапе добавление товара. Такую функциональность мы реализовали ранее, в рамках сценария админ панели.

Суть в том, что мы сохраняли не просто текст, а отформатированный текст:

await state.update_data(description=message.html_text)

В базе данных текст выглядит так:

То есть бот автоматически трансформирует обычный текст в текст с HTML тегами, а благодаря тому, что при инициализации объекта бота мы сделали это следующей строкой:

bot = Bot(token=settings.BOT_TOKEN,            default=DefaultBotProperties(parse_mode=ParseMode.HTML))

Мы по умолчанию установили трансформацию текста с HTML тегами в отформатированный текст.

Теперь мы приступаем к самому интересному — реализация оплаты в боте.

Подключение оплаты в боте

Триггером для запуска сценария оплаты в нашем телеграм-бот послужит клик на кнопку «Купить». После этого вызывается данная функция:

@catalog_router.callback_query(F.data.startswith('buy_')) async def process_about(call: CallbackQuery, session_without_commit: AsyncSession):     user_info = await UserDAO.find_one_or_none(         session=session_without_commit,         filters=TelegramIDModel(telegram_id=call.from_user.id)     )     _, product_id, price = call.data.split('_')     await bot.send_invoice(         chat_id=call.from_user.id,         title=f'Оплата 👉 {price}₽',         description=f'Пожалуйста, завершите оплату в размере {price}₽, чтобы открыть доступ к выбранному товару.',         payload=f"{user_info.id}_{product_id}",         provider_token=settings.PROVIDER_TOKEN,         currency='rub',         prices=[LabeledPrice(             label=f'Оплата {price}',             amount=int(price) * 100         )],         reply_markup=get_product_buy_kb(price)     )     await call.message.delete()  

И на этом месте давайте остановимся подробнее.

В библиотеке Aiogram 3 метод await bot.send_invoice используется для отправки пользователю инвойса (счета) на оплату через Telegram Payments. Этот функционал позволяет ботам работать с платежными системами, чтобы пользователи могли оплачивать товары или услуги прямо в чате с ботом.

Подробное описание метода await bot.send_invoice

Метод send_invoice — это асинхронная функция, которая отправляет сообщение с описанием товара, его ценой и кнопкой для оплаты. После нажатия на кнопку пользователь перенаправляется в интерфейс оплаты Telegram, где завершает транзакцию.

Основные параметры метода:

1. chat_id:

   • Идентификатор чата, куда будет отправлено сообщение.

   • Обычно используется call.from_user.id для отправки инвойса инициатору запроса.

2. title:

   • Заголовок инвойса.

   • Коротко описывает товар или услугу.

3. description:

   • Описание инвойса.

   • Например, можно указать детали покупки или напомнить, за что именно идет оплата.

4. payload:

   • Уникальный идентификатор заказа.

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

5. provider_token:

   • Токен платежного провайдера, выданный Telegram при настройке платежной системы.

6. currency:

   • Код валюты (например, 'rub' для рублей).

7. prices:

   • Список объектов LabeledPrice, описывающих стоимость товара.

     • label — Название позиции (например, «Оплата 500»).

     • amount — Сумма в минимальных единицах валюты (например, 500 рублей = 50000 копеек).

8. reply_markup:

   • Кастомная клавиатура или инлайн-кнопки, которые появятся вместе с инвойсом (опционально).

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

1. Пользователь нажимает кнопку «Купить», вызывая обработчик.

2. Бот отправляет инвойс с помощью send_invoice.

3. Telegram отображает инвойс с кнопкой «Оплатить».

4. Пользователь завершает ввод платежных данных.

5. Telegram отправляет вашему боту событие pre_checkout_query для подтверждения заказа. Бот должен обработать это событие в течение 10 секунд.

6. Если бот подтверждает запрос, транзакция завершается, и Telegram отправляет событие successful_payment.

Обработка pre_checkout_query

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

Пример простейшей обработки:

@catalog_router.pre_checkout_query(lambda query: True) async def pre_checkout_query(pre_checkout_q: PreCheckoutQuery):     await bot.answer_pre_checkout_query(pre_checkout_q.id, ok=True) 

• pre_checkout_q.id — Уникальный идентификатор запроса, который вы используете для ответа.

• ok=True — Подтверждает готовность завершить оплату.

• Если ok=False, транзакция будет отменена.

Здесь можно добавить логику проверки, например:

• Достаточно ли товара на складе?

• Совпадает ли цена товара с актуальной?

Обработка успешной оплаты (successful_payment)

После завершения транзакции Telegram отправляет событие successful_payment. Обработав его, вы можете сохранить данные об оплате в базе данных и предоставить пользователю доступ к купленному товару.

Пример реализации:

@catalog_router.message(F.content_type == ContentType.SUCCESSFUL_PAYMENT) async def successful_payment(message: Message, session_with_commit: AsyncSession):     payment_info = message.successful_payment     user_id, product_id = payment_info.invoice_payload.split('_')     payment_data = {         'user_id': int(user_id),         'payment_id': payment_info.telegram_payment_charge_id,         'price': payment_info.total_amount / 100,         'product_id': int(product_id)     }     # Добавляем информацию о покупке в базу данных     await PurchaseDao.add(session=session_with_commit, values=PaymentData(**payment_data))     product_data = await ProductDao.find_one_or_none_by_id(session=session_with_commit, data_id=int(product_id))      # Формируем уведомление администраторам     for admin_id in settings.ADMIN_IDS:         try:             username = message.from_user.username             user_info = f"@{username} ({message.from_user.id})" if username else f"c ID {message.from_user.id}"              await bot.send_message(                 chat_id=admin_id,                 text=(                     f"💲 Пользователь {user_info} купил товар <b>{product_data.name}</b> (ID: {product_id}) "                     f"за <b>{product_data.price} ₽</b>."                 )             )         except Exception as e:             logger.error(f"Ошибка при отправке уведомления администраторам: {e}")      # Подготавливаем текст для пользователя     file_text = "📦 <b>Товар включает файл:</b>" if product_data.file_id else "📄 <b>Товар не включает файлы:</b>"     product_text = (         f"🎉 <b>Спасибо за покупку!</b>\n\n"         f"🛒 <b>Информация о вашем товаре:</b>\n"         f"━━━━━━━━━━━━━━━━━━\n"         f"🔹 <b>Название:</b> <b>{product_data.name}</b>\n"         f"🔹 <b>Описание:</b>\n<i>{product_data.description}</i>\n"         f"🔹 <b>Цена:</b> <b>{product_data.price} ₽</b>\n"         f"🔹 <b>Закрытое описание:</b>\n<i>{product_data.hidden_content}</i>\n"         f"━━━━━━━━━━━━━━━━━━\n"         f"{file_text}\n\n"         f"ℹ️ <b>Информацию о всех ваших покупках вы можете найти в личном профиле.</b>"     )      # Отправляем информацию о товаре пользователю     if product_data.file_id:         await message.answer_document(             document=product_data.file_id,             caption=product_text,             reply_markup=main_user_kb(message.from_user.id)         )     else:         await message.answer(             text=product_text,             reply_markup=main_user_kb(message.from_user.id)         ) 

Итог

1. Метод send_invoice отправляет пользователю счёт.

2. Обработчик pre_checkout_query подтверждает готовность завершить оплату.

3. После события successful_payment бот сохраняет данные о транзакции и предоставляет товар пользователю.

Вот как данная реализация выглядит в боте:

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

Платёжный блок уже готов к работе. Чтобы начать принимать реальные платежи, вам нужно будет получить боевой токен через Telegram-бота Юкассы, установив его вместо тестового токена.

Опишем общую пользовательскую логику (user_router.py)

Выполним импорты и назначим роутер:

from aiogram import Router, F from aiogram.filters import CommandStart from aiogram.types import Message, CallbackQuery from sqlalchemy.ext.asyncio import AsyncSession from bot.dao.dao import UserDAO from bot.user.kbs import main_user_kb, purchases_kb from bot.user.schemas import TelegramIDModel, UserModel  user_router = Router() 

Опишем функцию, которая будет выполняться при первом входе пользователя в бота и при вводе команды /start:

@user_router.message(CommandStart()) async def cmd_start(message: Message, session_with_commit: AsyncSession):     user_id = message.from_user.id     user_info = await UserDAO.find_one_or_none(         session=session_with_commit,         filters=TelegramIDModel(telegram_id=user_id)     )      if user_info:         return await message.answer(             f"👋 Привет, {message.from_user.full_name}! Выберите необходимое действие",             reply_markup=main_user_kb(user_id)         )      values = UserModel(         telegram_id=user_id,         username=message.from_user.username,         first_name=message.from_user.first_name,         last_name=message.from_user.last_name,     )     await UserDAO.add(session=session_with_commit, values=values)     await message.answer(f"🎉 <b>Благодарим за регистрацию!</b>. Теперь выберите необходимое действие.",                          reply_markup=main_user_kb(user_id))  

Тут мы проверяем есть пользователь в базе данных или нет. Если есть, то выполняем его регистрацию, иначе просто отправляем ему сообщение. В виде-демке, которая будет представлена в конце статьи я продемонстрирую работу с новым пользователем.

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

@user_router.callback_query(F.data == "home") async def page_home(call: CallbackQuery):     await call.answer("Главная страница")     return await call.message.answer(         f"👋 Привет, {call.from_user.full_name}! Выберите необходимое действие",         reply_markup=main_user_kb(call.from_user.id)     )  

Вот простая функция с логикой «О нас»:

@user_router.callback_query(F.data == "about") async def page_about(call: CallbackQuery):     await call.answer("О магазине")     await call.message.answer(         text=(             "🎓 Добро пожаловать в наш учебный магазин!\n\n"             "🚀 Этот бот создан как демонстрационный проект для статьи на Хабре.\n\n"             "👨‍💻 Автор: Яковенко Алексей\n\n"             "🛍️ Здесь вы можете изучить принципы работы телеграм-магазина, "             "ознакомиться с функциональностью и механизмами взаимодействия с пользователем.\n\n"             "📚 Этот проект - это отличный способ погрузиться в мир разработки ботов "             "и электронной коммерции в Telegram.\n\n"             "💡 Исследуйте, учитесь и вдохновляйтесь!\n\n"             "Данные для тестовой оплаты:\n\n"             "Карта: 1111 1111 1111 1026\n"             "Годен до: 12/26\n"             "CVC-код: 000\n"         ),         reply_markup=call.message.reply_markup     )  

Далее опишем функцию, которая будет вызываться при клике пользователем на кнопку «Мои покупки».

@user_router.callback_query(F.data == "my_profile") async def page_about(call: CallbackQuery, session_without_commit: AsyncSession):     await call.answer("Профиль")      # Получаем статистику покупок пользователя     purchases = await UserDAO.get_purchase_statistics(session=session_without_commit, telegram_id=call.from_user.id)     total_amount = purchases.get("total_amount", 0)     total_purchases = purchases.get("total_purchases", 0)      # Формируем сообщение в зависимости от наличия покупок     if total_purchases == 0:         await call.message.answer(             text="🔍 <b>У вас пока нет покупок.</b>\n\n"                  "Откройте каталог и выберите что-нибудь интересное!",             reply_markup=main_user_kb(call.from_user.id)         )     else:         text = (             f"🛍 <b>Ваш профиль:</b>\n\n"             f"Количество покупок: <b>{total_purchases}</b>\n"             f"Общая сумма: <b>{total_amount}₽</b>\n\n"             "Хотите просмотреть детали ваших покупок?"         )         await call.message.answer(             text=text,             reply_markup=purchases_kb()         )  

Если коротко, то тут мы проверяем были ли покупки у пользователя и в зависимости от этого отправляем то или иное сообщение с той или иной клавиатурой.

Остается описать ту логику, которая будет вызвана при клике на «Смотреть покупки».

@user_router.callback_query(F.data == "purchases") async def page_user_purchases(call: CallbackQuery, session_without_commit: AsyncSession):     await call.answer("Мои покупки")      # Получаем список покупок пользователя     purchases = await UserDAO.get_purchased_products(session=session_without_commit, telegram_id=call.from_user.id)      if not purchases:         await call.message.edit_text(             text=f"🔍 <b>У вас пока нет покупок.</b>\n\n"                  f"Откройте каталог и выберите что-нибудь интересное!",             reply_markup=main_user_kb(call.from_user.id)         )         return      # Для каждой покупки отправляем информацию     for purchase in purchases:         product = purchase.product         file_text = "📦 <b>Товар включает файл:</b>" if product.file_id else "📄 <b>Товар не включает файлы:</b>"          product_text = (             f"🛒 <b>Информация о вашем товаре:</b>\n"             f"━━━━━━━━━━━━━━━━━━\n"             f"🔹 <b>Название:</b> <i>{product.name}</i>\n"             f"🔹 <b>Описание:</b>\n<i>{product.description}</i>\n"             f"🔹 <b>Цена:</b> <b>{product.price} ₽</b>\n"             f"🔹 <b>Закрытое описание:</b>\n<i>{product.hidden_content}</i>\n"             f"━━━━━━━━━━━━━━━━━━\n"             f"{file_text}\n"         )          if product.file_id:             # Отправляем файл с текстом             await call.message.answer_document(                 document=product.file_id,                 caption=product_text,             )         else:             # Отправляем только текст             await call.message.answer(                 text=product_text,             )      await call.message.answer(         text="🙏 Спасибо за доверие!",         reply_markup=main_user_kb(call.from_user.id)     )  

Теперь останется только настроить файл main.py и запустить бота.

Настройка main.py и запуск бота

Для того чтобы ваш бот начал работать, необходимо настроить главный файл проекта — bot/main.py. Этот файл отвечает за регистрацию роутеров, мидлварей и функций запуска и остановки бота. Здесь же происходит настройка командного меню и запуск бота в режиме long polling.

В реальных проектах лучше использовать подход с веб-хуками. В моём профиле на Хабре вы найдёте около пяти публикаций, где я на примере разных Telegram-ботов показывал, как запускать их через веб-хуки с помощью FastApi и Aiohttp.

Полный код файла main.py:

import asyncio from aiogram.types import BotCommand, BotCommandScopeDefault from loguru import logger from bot.config import bot, admins, dp from bot.dao.database_middleware import DatabaseMiddlewareWithoutCommit, DatabaseMiddlewareWithCommit from bot.admin.admin import admin_router from bot.user.user_router import user_router from bot.user.catalog_router import catalog_router  # Функция, которая настроит командное меню (дефолтное для всех пользователей) async def set_commands():     commands = [BotCommand(command='start', description='Старт')]     await bot.set_my_commands(commands, BotCommandScopeDefault())  # Функция, которая выполнится, когда бот запустится async def start_bot():     await set_commands()     for admin_id in admins:         try:             await bot.send_message(admin_id, f'Я запущен🥳.')         except:             pass     logger.info("Бот успешно запущен.")  # Функция, которая выполнится, когда бот завершит свою работу async def stop_bot():     try:         for admin_id in admins:             await bot.send_message(admin_id, 'Бот остановлен. За что?😔')     except:         pass     logger.error("Бот остановлен!")  async def main():     # Регистрация мидлварей     dp.update.middleware.register(DatabaseMiddlewareWithoutCommit())     dp.update.middleware.register(DatabaseMiddlewareWithCommit())      # Регистрация роутеров     dp.include_router(catalog_router)     dp.include_router(user_router)     dp.include_router(admin_router)      # Регистрация функций     dp.startup.register(start_bot)     dp.shutdown.register(stop_bot)      # Запуск бота в режиме long polling     try:         await bot.delete_webhook(drop_pending_updates=True)         await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())     finally:         await bot.session.close()  if __name__ == "__main__":     asyncio.run(main()) 

Разберем ключевые части кода:

1. Настройка командного меню

async def set_commands():     commands = [BotCommand(command='start', description='Старт')]     await bot.set_my_commands(commands, BotCommandScopeDefault()) 

Функция задает команды для бота. В данном случае добавляется команда /start с описанием «Старт».

2. Действия при запуске бота

async def start_bot():     await set_commands()     for admin_id in admins:         try:             await bot.send_message(admin_id, f'Я запущен🥳.')         except:             pass     logger.info("Бот успешно запущен.") 

При старте бота:

• Настраиваются команды с помощью set_commands.

• Отправляется уведомление администраторам.

• В логах фиксируется сообщение о запуске.

3. Действия при остановке бота

async def stop_bot():     try:         for admin_id in admins:             await bot.send_message(admin_id, 'Бот остановлен. За что?😔')     except:         pass     logger.error("Бот остановлен!") 

При остановке бота:

• Отправляется уведомление администраторам о завершении работы.

• В логах фиксируется сообщение об остановке.

4. Регистрация мидлварей и роутеров

async def main():     dp.update.middleware.register(DatabaseMiddlewareWithoutCommit())     dp.update.middleware.register(DatabaseMiddlewareWithCommit())      dp.include_router(catalog_router)     dp.include_router(user_router)     dp.include_router(admin_router) 

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

• Роутеры — это маршрутизаторы, которые группируют обработчики команд и событий.

5. Запуск бота

try:     await bot.delete_webhook(drop_pending_updates=True)     await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types()) finally:     await bot.session.close() 

• delete_webhook — очищает вебхуки и удаляет отложенные обновления. Эта запись в Aiogram 3, будет одинаково корректно работать, как при использовании поллинга, так и при использовании веб-хуков.

• start_polling — запускает бота в режиме long polling. Бот начинает получать обновления с серверов Telegram.

Теперь ваш бот готов к работе! Остается запустить main.py файл и протестировать его функционал.

Для запуска бота в консоли, с корня проекта, вводим следующую команду:

python -m bot.main

Ниже вы можете увидеть видео, на котором показан процесс работы бота как с пользовательской стороны, так и с административной.

Остается последний шаг, чтоб можно было уверенно сказать, что проект готов — деплой бота на сервис Amvera Cloud.

Подготовка и развертывание бота в облачном сервисе Amvera Cloud

Перед запуском бота в облачной среде Amvera Cloud необходимо провести ряд подготовительных мероприятий. Ключевым элементом этого процесса является создание конфигурационного файла amvera.yml в корневой директории проекта. Этот файл содержит важные инструкции, позволяющие Amvera Cloud корректно развернуть и запустить нашего бота.

Структура файла amvera.yml

meta:   environment: python   toolchain:     name: pip     version: 3.12 build:   requirementsPath: requirements.txt run:   persistenceMount: /data   containerPort: 8000   command: python3 -m bot.main 

Данный конфигурационный файл предоставляет Amvera следующую информацию:

• Проект написан на Python версии 3.12

• Зависимости должны быть установлены из файла requirements.txt с помощью pip

• Определяет место хранения базы данных

• Указывает команду для запуска бота

Особое внимание к блоку ‘run’

run:   persistenceMount: /data   command: python3 -m bot.main 

Строка persistenceMount: /data имеет критическое значение. Она указывает, что важные файлы, включая базу данных SQLite, должны быть примонтированы в папке /data. Это обеспечивает сохранность данных при пересборке проекта, которая может потребоваться при обновлении кода.

После развертывания крайне важно убедиться, что база данных действительно находится в папке /data. Это гарантирует, что данные не будут потеряны при обновлениях и пересборках проекта.

Запуск бота

Для запуска бота используется команда python3 -m bot.main. Важно отметить, что в отличие от Windows, где может использоваться команда python, в среде UNIX (Linux) необходимо использовать python3. Это обеспечивает корректный запуск бота в облачной среде Amvera Cloud, которая базируется на UNIX-системе.

Процесс деплоя в Amvera Cloud

1. Регистрация: Если у вас еще нет аккаунта, зарегистрируйтесь в Amvera Cloud. Новые пользователи получают приветственный бонус в размере 111 рублей на счет.

2. Создание проекта: В панели управления перейдите в раздел проектов и нажмите «Создать проект».

3. Настройка проекта: Присвойте проекту название и выберите подходящий тарифный план. Для учебных целей «Начальный» тариф будет оптимальным выбором.

4. Загрузка файлов: Выберите способ загрузки файлов приложения. Новичкам рекомендуется использовать интерфейс Amvera, более опытным пользователям — команды GIT.

5. Проверка настроек: На следующем экране убедитесь, что все настройки подгрузились корректно, и нажмите «Завершить».

6. Ожидание развертывания: Процесс деплоя займет 3-4 минуты. По завершении вы получите уведомление от бота о его успешном запуске, а в интерфейсе Amvera появится индикатор активного состояния приложения.

Кому интересно готового бота можно поклацать тут: https://t.me/DigitalMarketAiogramBot

Напоминаю, что полный исходный код сегодняшнего проекта и эксклюзивный контент, который я не публикую на Хабре вы найдете в моем бесплатно телеграмм канале «Легкий путь в Python».

Заключение

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

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

Чему мы научились:

• Разработка админки для Telegram-бота: Мы создали удобный интерфейс для управления ботом.

• Интеграция SQLAlchemy 2: Научились использовать эту мощную ORM для работы с данными в телеграмм ботах.

• Работа с машиной состояний: Освоили управление состояниями в боте, что позволяет создавать интерактивные сценарии.

• Интеграция платежной системы: Пройдя все этапы — от выставления счета до обработки успешного платежа и отправки уведомлений в бота.

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

Если эта статья была для вас полезной — дайте знать лайком или комментарием! Порой работа, над которой я трудился несколько дней, может остаться незамеченной, и в такие моменты я задумываюсь о том, что это все никому не нужно.

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

На этом у меня все. До скорого!