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="