Telegram Bot Store on Python: Step-by-Step Guide with Payment, Catalog, and Admin Panel (Aiogram 3 + SQLAlchemy 2)

Friends, hello! Today I come to you with a new practical project in Python. This time we will create a full-fledged Telegram bot for selling digital goods with a database that we will manage via SQLAlchemy 2, an admin panel, a user part and integrated payment via Yukasa.

I will take you step by step through all the stages of developing such a bot: from registering a token in BotFather to deploying the finished product on a remote hosting so that the bot can work smoothly 24/7 without being tied to your computer or Internet connection. But first things first.

What technologies will we use?

In this project, we will work exclusively with Python, so those who do not like JavaScript can relax today. We use the following technologies:

• Aiogram 3 — the best asynchronous Python framework for developing Telegram bots.

• SQLAlchemy 2 — a powerful Python ORM that supports asynchronous interaction with all tabular databases. We will work with SQLite, but the code written is also suitable for other databases, such as PostgreSQL.

• Aiosqlite — an asynchronous engine for interacting with SQLite via SQLAlchemy 2.

• Pydantic 2 — for data validation and working with settings.

• Alembic — for automating work with table structures (migrations).

Important: disclaimer!

We are developing a digital goods store, and I will show the "classic" payment connection in the bot via BotFather. However, since June 12, 2024, there is an obligation for Telegram bots selling digital goods to use the "Telegram Stars" payment system (stars).

From my experience, at the moment I have not encountered blocking of bots using classic payment. Dozens of my clients successfully use such bots, both before and after the specified date. Even projects completed in the last month work without problems. However, it is important to realize that the risks of blocking still exist.

In future publications, I will tell you how to work with "stars" in the Russian Federation. But for now we hope that blocking will not affect us, and we continue.

Step-by-step plan

As part of the project, we will complete the following stages:

1. Preparing a bot token and obtaining a test token for payment via Yukassa

• Creating a bot via BotFather.

• Obtaining a Yukassa payment token (we will consider how to obtain a test and "combat" token).

2. Writing the bot code

We will divide this stage into several subtasks:

• Describing database tables (creating models).

• Database migrations (converting models into real SQLite tables).

• Writing methods for interacting with the database (adding, deleting, changing and retrieving data).

• Implementing user logic (catalog, user profile, payment, "about us" information).

• Implementing administrative logic (statistics, adding and deleting products).

3. Bot deployment

• Writing code is not everything. To make the bot work 24/7, we will host it on a remote server.

• For this, we will use the Amverum Cloud service — a convenient domestic analogue of Heroku.

Deployment steps:

1. Preparing a file with settings (the code will be provided in the article).

2. Creating a project in Amverum.

3. Uploading bot files and settings (manually on the website or via GIT).

4. Waiting 3-4 minutes for the bot to build and run.

The process will take no more than 10 minutes and will be understandable even for beginners.

Setting up the bot and connecting payment

Before starting development, we need to get two tokens: the token of the bot itself and the test payment token. These keys will allow us to create the basis for the future digital goods store, which we will fill with code and functionality.

To complete this step, we will use the BotFather bot and the Yukassa payment system. As a result, you will get a full-fledged Telegram bot with test payment integration, and later you will be able to connect real payments.

Creating a bot via BotFather

Let's start by getting a bot token:

1. Open BotFather in Telegram.

2. Enter the command /newbot.

3. Specify the bot name (it can be in Russian).

4. Come up with a unique login for the bot in Latin, ending in BOT, bot or Bot.

Example

After completing these steps, you will receive a bot token - an important tool that will connect your code to the Telegram API. With this token, we will be able to connect the payment in the bot.

Linking the payment system to the bot

Now let's move on to setting up the acceptance of test payments.

1. Log in to your bot via BotFather.

2. Select the Payments section.

1. Click on Yukassa and select the Connect Test Yukassa option.

2. BotFather will redirect you to the Yukassa bot.

Inside the Yukassa bot:

• If you already have an account in Yukassa, select Log in and grant access.

• If you do not have an account, click Connect Yukassa and follow the instructions to register.

To activate test payments, make sure that you have a test store in your Yukassa account. This is a prerequisite for successful linking.

After successful connection, you will see the message:

3. Getting a test token

To get a test payment token:

1. Return to BotFather.

2. Open your bot settings and select Payments.

3. In the payments section, you will find your test token.

After completing all the steps, you should have two tokens on hand:

• Bot token — for working with the Telegram API.

• Test payment token — for integration with Yukassa.

Now we are ready for the next steps of development!

Project Organization

Let's move on to preparing the structure and starting to develop a Telegram bot for a digital goods store. Let's start with the general project setup and organizing the file structure.

Step 1: Create a project and set up the environment

1. In your favorite IDE, create a new project for the bot.

2. Activate the virtual environment to isolate dependencies.

After that, let's start creating the basic project structure:

project  
│  
├── bot/                # Bot's main code
│   ├── admin/          # Admin panel logic 
│   ├── user/           # User part logic 
│   ├── dao/            # Working with the database  
│   ├── config.py       # Project Settings  
│   └── main.py         # Main application file 
├── data/               # Database storage 
├── .env                # Environment variables (tokens, settings) 
└── requirements.txt    # Project dependencies 

Step 2: Description of the bot folder structure

The bot folder will contain all the main code of the bot. It includes:

• admin/ is a directory for the code related to the bot admin panel.

• user/ is a directory where we will place the logic of the user part of the bot, including commands, menus, and other interactions.

• dao/ is a module for working with the database. Models and methods for interacting with data will be stored here.

• config.py is a project settings file where we will specify basic parameters and paths.

• main.py is the main file from which the application is launched.

Step 3: Additional files and directories

• data/ is a folder for storing the database.

• .env is a file for environment variables, such as the bot token, payment system token, and other confidential data.

• requirements.txt — a list of all project dependencies for quick installation.

Step 4: Future changes to the project

Later, as we work on the bot, the project will have:

• Database migrations — we will place them in a separate folder.

• File alembic.ini — for managing migrations using Alembic.

• Files in microservices (created empty folders)

Project setup and dependency installation

At this stage, we will install the necessary libraries and configure the configuration files so that the project is ready for further development.

Installing dependencies

Add the following libraries to the requirements.txt file:

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

These libraries provide the basic functions of the bot, including working with the Telegram API, database, logging and settings management. Let's install them with the command:

pip install -r requirements.txt

Setting up the .env file

Create a .env file in the root of the project and fill it with the following variables:

BOT_TOKEN=YOUR_BOT_TOKEN
ADMIN_IDS=[AdminID1, AdminID2, AdminID3]
PROVIDER_TOKEN=PAYMENT_TOKEN

• BOT_TOKEN — your bot's token, which you received via BotFather.

• ADMIN_IDS — a list of Telegram IDs of administrators who will have access to the admin panel. To get the ID, you can use the IDBot Finder Pro bot.

• PROVIDER_TOKEN — the payment system token that we connected earlier.

Settings file: bot/config.py

In this file we will collect all the key project settings, including environment variables, logging parameters, database connection and bot initialization.

config.py file code:

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

Getting parameters for loading environment variables

settings = Settings()

Initialize the bot and dispatcher

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

Code parsing

1. Imports

• Libraries for working with environment variables (BaseSettings), logging (loguru), Telegram API (aiogram) and the database are used.

1. Class Settings

• Inherits from BaseSettings from pydantic-settings to automatically load environment variables from the .env file.

• Key parameters:

• BOT_TOKEN, ADMIN_IDS, PROVIDER_TOKEN are mandatory variables.

• FORMAT_LOG and LOG_ROTATION are logging settings.

• DB_URL is the URL for connecting to the SQLite database via aiosqlite.

• model_config specifies the path to the .env file.

1. Initializing settings

• The settings object is created, which loads variables from .env and provides access to them via class attributes.

1. Initializing the bot and dispatcher

• Bot and Dispatcher are initialized based on the token from settings.BOT_TOKEN.

• MemoryStorage is used to store FSM states in memory (in production projects it is better to use RedisStorage - described in detail in this article).

• admins contains a list of administrator IDs from the ADMIN_IDS variable.

1. Logging

• Using loguru, a log file log.txt is created, into which events are recorded with rotation when 10 MB are reached.

1. Database URL

• The database_url variable contains the path to connect to the SQLite database.

The project is now ready for the next stage of development - implementing the basic logic.

Database interaction logic

Now that the basic structure of the project is ready, we can start developing one of the most important parts of our Telegram bot - the database interaction logic.

Useful resources for working with SQLAlchemy 2 and Alembic

Before diving into the code, I recommend reading my articles that will help you better understand working with SQLAlchemy 2 and Alembic:

1. Asynchronous SQLAlchemy 2: a simple step-by-step guide to setting up, models, relationships, and migrations using Alembic.

2. Asynchronous SQLAlchemy 2: a step-by-step guide to managing sessions, adding and retrieving data with Pydantic.

3. Asynchronous SQLAlchemy 2: improving code, update and delete methods data.

In these articles you will find detailed information about working with models, relationships and migrations. Further I will assume that you are already familiar with the basics or have studied them from the specified materials.

bot/dao folder structure

All database related code will be in the bot/dao folder. Create it with the following structure:

├── dao/                
│   ├── __init__.py                # Batch file for easy imports
│   ├── database.py                # SQLAlchemy settings
│   ├── models.py                  # Database models
│   ├── base.py                    # Universal class for interaction with the database
│   ├── dao.py                     # Specialized DAO classes
│   └── database_middleware.py     # Middleware for managing database sessions

File database.py

The database.py file is responsible for configuring SQLAlchemy and creating a base class for all models.

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

# Creating an asynchronous engine to connect to the database
engine = create_async_engine(url=database_url)

# Creating a session factory
async_session_maker = async_sessionmaker(engine, class_=AsyncSession)

# Base class for models
class Base(AsyncAttrs, DeclarativeBase):
    __abstract__ = True  # This class will not create a separate table

    # Common field "id" for all tables
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

    # Record creation and update time fields
    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()
    )

    # Automatically detect table name
    @classmethod
    @property
    def __tablename__(cls) -> str:
        return cls.__name__.lower() + 's'

Brief analysis of the file:

• engine: Asynchronous engine for working with the database. Connected using create_async_engine.

• async_session_maker: Factory for creating asynchronous sessions.

• Class Base:

• Base abstract class for all models.

• Field id: Common primary key.

• Fields created_at and updated_at: Automatic management of the time of creation and update of records.

File models.py

This file describes the main database models: users, categories, products and purchases.

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})>"

Key points:

• User: Describes a Telegram user. Contains links to purchases via relationship.

• Category: Product category. Contains links to products.

• Product: Describes a product with the fields: name, description, price, file, and hidden content.

• Purchase: Purchase information with the user, product, and price.

Now we are ready to transform the database models into a full-fledged SQLite database with tables that will contain the internal relationships we need. For this, we will use the Alembic tool.

Setting up Alembic and creating the first migration

To effectively manage changes to the database, let's set up Alembic and create the first migration. This will automatically track changes to the models and update the database structure.

Initializing Alembic

First, go to the bot directory:

cd bot

Initialize Alembic with asynchronous database support:

alembic init -t async migration

After executing the command, the migration folder and the alembic.ini file will appear. Move alembic.ini to the root directory of the project for ease of use.

Setting up the alembic.ini file

Open the alembic.ini file and change the line:

script_location = migration

On:

script_location = bot/migration

This makes it easier to use migrations and run a project from the root directory.

Modifying env.py to connect to the database

Now we need to make changes to the bot/migration/env.py file so that Alembic can work correctly with our database. Open the file and replace its contents with the following:

Was

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

It became

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

The rest of the file can be left unchanged. In the future, added table models will be automatically tracked.

Key changes:

1. Adding the project root directory to the system path for correct import of modules.

2. Specifying the URL for connecting to the database.

3. Setting up metadata for automatic generation of migrations.

Creating the first migration

Go to the root directory of the project:

cd ../

Generate the migration file:

alembic revision --autogenerate -m "Initial revision"

Apply migrations to create tables in the database:

alembic upgrade head

After running this command, a file db.sqlite3 will appear in the root of the project, containing the tables users, purchases, products, and categories.

Working with categories

You can manually add data to fill the category table. The id, created_at, and updated_at columns are automatically filled in for each record. You only need to specify category_name.

As an exercise, I suggest implementing the logic for managing categories through a bot yourself, since this block will not be covered in the article.

After "manually" adding categories, I got the following result:

In the finished project, categories will be automatically loaded into the inline keyboard, which will make the interface user-friendly.

This is how the catalog page implementation will look in our bot after integrating the category table into it.

Now we can proceed to the methods for interacting with the database.

Writing logic for universal work with the database (BaseDao class)

To simplify interaction with the database and reduce code duplication, the project uses the universal BaseDAO class. This class provides basic methods that can be overridden and supplemented in child classes.

The BaseDAO class is based on template programming and allows you to work with any model inherited from Base.

The class itself in the source code of the project is larger than will be discussed below. Therefore, if you want to get the full code of this class, I invite you to my free telegram channel "Easy way to Python". There you will find not only the full source code of today's project, but also material that I do not publish on Habr.

In the meantime, we will describe only those methods that we will use in our project:

You will get the following code:

Hidden text

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

# We declare a type parameter T with the constraint that it is a descendant of 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):
        # Find a record by 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} found.")
            else:
                logger.info(f"Запись с ID {data_id} not found.")
            return record
        except SQLAlchemyError as e:
            logger.error(f"Error searching for record with ID {data_id}: {e}")
            raise

    @classmethod
    async def find_one_or_none(cls, session: AsyncSession, filters: BaseModel):
        # Find one entry by filters
        filter_dict = filters.model_dump(exclude_unset=True)
        logger.info(f"Search for a single entry {cls.model.__name__} by filters: {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"Record found by filters: {filter_dict}")
            else:
                logger.info(f"Record not found by filters: {filter_dict}")
            return record
        except SQLAlchemyError as e:
            logger.error(f"Error while searching record by filters {filter_dict}: {e}")
            raise

    @classmethod
    async def find_all(cls, session: AsyncSession, filters: BaseModel | None = None):
        # Find all records by filters
        filter_dict = filters.model_dump(exclude_unset=True) if filters else {}
        logger.info(f"Search all entries {cls.model.__name__} by filters: {filter_dict}")
        try:
            query = select(cls.model).filter_by(**filter_dict)
            result = await session.execute(query)
            records = result.scalars().all()
            logger.info(f"Found {len(records)} records.")
            return records
        except SQLAlchemyError as e:
            logger.error(f"Error while searching all records by filters {filter_dict}: {e}")
            raise

    @classmethod
    async def add(cls, session: AsyncSession, values: BaseModel):
        # Add one entry
        values_dict = values.model_dump(exclude_unset=True)
        logger.info(f"Adding a record {cls.model.__name__} with parameters: {values_dict}")
        new_instance = cls.model(**values_dict)
        session.add(new_instance)
        try:
            await session.flush()
            logger.info(f"Recording {cls.model.__name__} successfully added.")
        except SQLAlchemyError as e:
            await session.rollback()
            logger.error(f"Error adding record: {e}")
            raise e
        return new_instance

    @classmethod
    async def delete(cls, session: AsyncSession, filters: BaseModel):
        # Delete records by filter
        filter_dict = filters.model_dump(exclude_unset=True)
        logger.info(f"Deleting records {cls.model.__name__} by filter: {filter_dict}")
        if not filter_dict:
            logger.error("At least one filter is needed to remove.")
            raise ValueError("At least one filter is needed to remove.")

        query = sqlalchemy_delete(cls.model).filter_by(**filter_dict)
        try:
            result = await session.execute(query)
            await session.flush()
            logger.info(f"Removed {result.rowcount} records.")
            return result.rowcount
        except SQLAlchemyError as e:
            await session.rollback()
            logger.error(f"Error deleting records: {e}")
            raise e

    @classmethod
    async def count(cls, session: AsyncSession, filters: BaseModel | None = None):
        # Count the number of records
        filter_dict = filters.model_dump(exclude_unset=True) if filters else {}
        logger.info(f"Counting the number of records {cls.model.__name__} by filter: {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"Found {count} records.")
            return count
        except SQLAlchemyError as e:
            logger.error(f"Error while counting records: {e}")
            raise

In my articles on SQLAlchemy I examined the principles of interaction with the BaseDao class in more detail.

Using BaseDAO

For each model, a child class is created, inherited from BaseDAO. For example:

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

This allows you to call methods directly, for example:

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

If the basic methods are not enough, you can add your own methods in the child class.

Important point

BaseDAO methods intentionally do not commit changes to the database. This allows you to perform several operations within a single session and fix them with a single commit, if necessary. This approach is especially convenient in asynchronous projects, for example, in Telegram bots.

How does it work in a Telegram bot?

1. The user initiates an action through the bot.

2. The middleware creates and opens a session.

3. The necessary operations with the database are performed in the processing function.

4. After the operation is completed, the session is automatically closed and, if necessary, performs a commit.

This approach will be discussed in detail in practice below.

Child DAO classes: working with specific models

The project uses child classes of BaseDAO to manage data. Each of them is tied to a specific model and can contain additional methods if required. In this section, we will analyze the implementation of such classes in the bot/dao/dao.py file.

Imports

First, let's connect the necessary modules and libraries:

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

Simple child classes

For some models it is enough to simply inherit from BaseDAO and specify the corresponding model. This allows you to use all the basic BaseDAO methods, such as adding, deleting, finding and counting records.

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


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

These classes can already be used without adding additional methods.

Working with purchases: PurchaseDao

The purchase model requires implementing a method to calculate the total amount of all purchases. This is useful, for example, for collecting statistics in the admin panel.

class PurchaseDao(BaseDAO[Purchase]):
    model = Purchase

    @classmethod
    async def get_full_summ(cls, session: AsyncSession) -> int:
        """Get the total amount of purchases."""
        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

This method returns the total price of all purchases. If there are no purchases in the database, the method returns 0.

Working with users: UserDAO

The user model requires the implementation of several additional methods to obtain statistics and related data. Let's consider them in detail.

User purchase statistics

This method returns the total number of purchases and their amount for a specific user by their 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:
            # Query to get the total number of purchases and the total amount
            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  # Handling the case where the amount can be None
            }

        except SQLAlchemyError as e:
            # Handling errors when working with a database
            print(f"Error getting user purchase statistics: {e}")
            return None

User Shopping List

The method returns a list of all user purchases with product details.

   @classmethod
    async def get_purchased_products(cls, session: AsyncSession, telegram_id: int) -> Optional[List[Purchase]]:
        try:
            # Query to get a user with their purchases and related products
            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:
            # Handling errors when working with a database
            print(f"Error retrieving user purchase information: {e}")
            return None

This method loads related purchases and products via ORM loading (selectinload) to minimize the number of database queries.

General User Statistics

This method collects data on the number of users registered for different time periods.

   @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 received successfully: {statistics}")
            return statistics
        except SQLAlchemyError as e:
            logger.error(f"Error while getting statistics: {e}")
            raise

The method calculates the number of users registered over the last day, week, month, and the total number of users.

Now we have the last step in setting up interaction with the database via SQLAlchemy, namely, describing the middleware.

Creating a middleware for managing database sessions

What is a middleware?

Middleware in Telegram bots is an intermediate layer of logic that is executed between receiving an event (for example, a message or callback request) and processing it by a handler. They allow you to change event data, add additional parameters or perform third-party actions (for example, creating a session for working with a database).

Why do we need middleware for session management?

In our project, we implement automatic management of database sessions. This means that:

• The session is automatically opened before processing the event.

• Depending on the need to change the data, the session is either committed or rolled back.

• After the processing is complete, the session is automatically closed.

For this, we will create two middlewares: one for managing the session without commit, the other - with commit. To avoid duplicating code, we will implement a common base class, and then create two child classes.

Working with the file bot/dao/database_middleware.py

Let's start with the necessary imports:

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

Base class implementation

We will place the core of the session management logic in the BaseDatabaseMiddleware base class. It will:

1. Open a session before handling an event.

2. Pass the session to data — a special dictionary used in Aiogram to pass data between middlewares and handlers.

3. Close the session automatically, regardless of the outcome of the processing.

This is what this class looks like:

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)  # Setting up a session
            try:
                result = await handler(event, data)  # Processing the event
                await self.after_handler(session)  # Additional actions (eg commit)
                return result
            except Exception as e:
                await session.rollback()  # Rollback changes in case of error
                raise e
            finally:
                await session.close()  # Closing the session

    def set_session(self, data: Dict[str, Any], session) -> None:
        """Method for setting session to data. Implemented in child classes."""
        raise NotImplementedError("This method must be implemented in subclasses..")

    async def after_handler(self, session) -> None:
        """Method to perform actions after handling an event. Does nothing by default."""
        pass

Child classes for session management

Now let's create two child classes with concrete implementation of logic:

1. Middleware for session without commit

This middleware simply passes the session to data without committing.

class DatabaseMiddlewareWithoutCommit(BaseDatabaseMiddleware):
    def set_session(self, data: Dict[str, Any], session) -> None:
        """Setting up a session without committing."""
        data['session_without_commit'] = session

1. Middleware for session with commit

This middleware additionally commits changes to the database after successful event processing.

class DatabaseMiddlewareWithCommit(BaseDatabaseMiddleware):
    def set_session(self, data: Dict[str, Any], session) -> None:
        """Set up a session with a commit."""
        data['session_with_commit'] = session

    async def after_handler(self, session) -> None:
        """We record changes after event processing."""
        await session.commit()

How does it work?

Each of the middlewares automatically:

1. Opens a session via async_session_maker().

2. Adds the session to the data for subsequent use in handlers.

3. When processing is complete:

• If DatabaseMiddlewareWithCommit is used, a commit is performed.

• If DatabaseMiddlewareWithoutCommit is used, changes to the database remain unprocessed.

1. In case of an error, rolls back the changes and closes the session.

Next, in order for our bot to see the middlewares, we only need to register them. We will not dwell on this now, since later, using a practical example, we will analyze both the registration and use of middlewares in the bot functions.

This closes the block on setting up the logic of interaction with the database, which means that we can move on to the parts of the bot.

Writing the bot admin panel

Let's start with the description of the admin panel. This is logical, since the admin panel will contain the functionality for adding products to our store. Without products, the store does not exist, so we will implement this part first.

Admin panel file structure

We will work with the bot/admin folder, in which we will create the following structure:

├── admin/                
│   ├── __init__.py           # Batch file for easy imports
│   ├── admin.py              # Main file with admin panel methods
│   ├── kbs.py                # Description of admin panel keyboards
│   ├── schemas.py            # Pydantic schemas for working with data
│   └── utils.py              # Auxiliary utilities for the admin panel

This structure is compact and convenient for small projects. In large projects, you can split the logic into additional folders. For example, I often place dao.py and models.py files in such modules, where I describe specific models and child DAO classes so that the module can be scaled. A similar approach is also used when creating microservices on FastAPI.

utils.py file

Let's start with utilities. Here we will implement a method for safely deleting previous messages. This will help to avoid cluttering the administrator's dialogue with the bot. The method will look like this:

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("Error: Could not find last message id to delete.")
        await message.delete()

    except Exception as e:
        logger.error(f"There was an error deleting the message: {str(e)}")

Method explanation:

• We pass the message object and the current state to the method.

• The state is used to get the ID of the last message (last_msg_id) that needs to be deleted.

• The current message is deleted via the await message.delete() method.

• If something went wrong, we log the error.

Schemas.py file

Here we will describe the Pydantic schemas that are used to validate and add data to the tables associated with the admin panel. Let me remind you that in BaseDAO we use instances of the Pydantic BaseModel as arguments for adding or filtering data, not dictionaries.

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)

Explanation:

• ProductIDModel is used to pass the product ID.

• ProductModel describes the data structure for the product, including the name, description, price, category ID, file, and hidden content.

• Field parameters specify additional constraints, such as a minimum string length or a positive price value.

kbs.py file

Now let's create a file with keyboards. All buttons will be implemented using InlineKeyboardBuilder, which provides a convenient way to work with inline buttons in 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="Cancel", callback_data="admin_panel")
    kb.adjust(2)
    return kb.as_markup()


def admin_send_file_kb() -> InlineKeyboardMarkup:
    kb = InlineKeyboardBuilder()
    kb.button(text="No file", callback_data="without_file")
    kb.button(text="Cancel", callback_data="admin_panel")
    kb.adjust(2)
    return kb.as_markup()


def admin_kb() -> InlineKeyboardMarkup:
    kb = InlineKeyboardBuilder()
    kb.button(text="📊 Statistics", callback_data="statistic")
    kb.button(text="🛍️ Manage products", callback_data="process_products")
    kb.button(text="🏠 To the main page", callback_data="home")
    kb.adjust(2)
    return kb.as_markup()


def admin_kb_back() -> InlineKeyboardMarkup:
    kb = InlineKeyboardBuilder()
    kb.button(text="⚙️ Admin panel", callback_data="admin_panel")
    kb.button(text="🏠 To the main page", callback_data="home")
    kb.adjust(1)
    return kb.as_markup()


def dell_product_kb(product_id: int) -> InlineKeyboardMarkup:
    kb = InlineKeyboardBuilder()
    kb.button(text="🗑️ Delete", callback_data=f"dell_{product_id}")
    kb.button(text="⚙️ Admin panel", callback_data="admin_panel")
    kb.button(text="🏠 To the main page", callback_data="home")
    kb.adjust(2, 2, 1)
    return kb.as_markup()


def product_management_kb() -> InlineKeyboardMarkup:
    kb = InlineKeyboardBuilder()
    kb.button(text="➕ Add product", callback_data="add_product")
    kb.button(text="🗑️ Delete product", callback_data="delete_product")
    kb.button(text="⚙️ Admin panel", callback_data="admin_panel")
    kb.button(text="🏠 To the main page", callback_data="home")
    kb.adjust(2, 2, 1)
    return kb.as_markup()


def cancel_kb_inline() -> InlineKeyboardMarkup:
    kb = InlineKeyboardBuilder()
    kb.button(text="Cancel", callback_data="cancel")
    return kb.as_markup()


def admin_confirm_kb() -> InlineKeyboardMarkup:
    kb = InlineKeyboardBuilder()
    kb.button(text="Everything is correct", callback_data="confirm_add")
    kb.button(text="Cancel", callback_data="admin_panel")
    kb.adjust(1)
    return kb.as_markup()

Explanation:

• All keyboards are designed as functions that return the InlineKeyboardMarkup object.

• We use modern methods from Aiogram 3 to dynamically create buttons.

• Buttons are grouped 2-3 in a row for ease of display.

Now we can start writing the main admin panel code.

Describing the admin panel logic

Before we start describing the code, I want to assume that you are already familiar with the Aiogram 3 framework. If this is not the case, then I recommend that you read my articles, in which I talk in detail, from start to finish, about creating Telegram bots based on Aiogram 3. You can find all the articles in chronological order at the following link: Full list of articles.

Next, I will assume that you have basic knowledge of interaction with Aiogram 3.

Now we are ready to describe the main logic of the admin panel. We will write the code in the file bot/admin/admin.py

Let's start with imports and initialization of the router:

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()

Now let's describe the class in which we will store our FSM states:

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

Now let's start with the implementation of individual methods. I will only dwell in detail on new technologies that I did not consider in previous articles, otherwise this publication will take a couple of hours to read, which is not necessary for me or you.

Let's describe the first function that will be executed when you log in to the admin panel:

@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('Access to the admin panel is allowed!')
    await call.message.edit_text(
        text="You are allowed to access the admin panel. Select the required action.",
        reply_markup=admin_kb()
    )

Here I used Aiogram 3's magic filters to verify that the method was called by the administrator:

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

And to check that call_data – "admin_panel" was called:

F.data == "admin_panel"

We take the list of telegram IDs of administrators with the variable ADMIN_IDS.

Next, if the checks are successful, the bot will send the user the main keyboard of the inline panel, otherwise nothing will happen, since we did not write a separate handler for non-admins.

Let's describe the method that will return statistics for the bot.

@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('Request for statistics...')
    await call.answer('📊 We collect statistics...')

    stats = await UserDAO.get_statistics(session=session_without_commit)
    total_summ = await PurchaseDao.get_full_summ(session=session_without_commit)
    stats_message = (
        "📈 User statistics:\n\n"
        f"👥 Total users: {stats['total_users']}\n"
        f"🆕 New today: {stats['new_today']}\n"
        f"📅 New this week: {stats['new_week']}\n"
        f"📆 New for the month: {stats['new_month']}\n\n"
        f"💰 Total amount of orders: {total_summ} usd.\n\n»
        "🕒 The data is current as of now."
    )
    await call.message.edit_text(
        text=stats_message,
        reply_markup=admin_kb()
    )

What's noteworthy here is the session call:

session_without_commit: AsyncSession

It is important here that the name of the variable with the session matches the name we set in the middleware. In this context, we are interested in a session without a commit, since we are simply receiving data from the database. That is, we do not need to record anything.

Otherwise, we simply call pre-prepared DAO methods, and then format the received information into beautiful text and send it to the administrator.

Now let's analyze the main page of the block for managing goods. It looks like this:

@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('Product Management Mode')
    all_products_count = await ProductDao.count(session=session_without_commit)
    await call.message.edit_text(
        text=f"Currently in the database {all_products_count} goods. What are we going to do?",
        reply_markup=product_management_kb()
    )

What's noteworthy here is that for ProductDao we called the count method, which was inherited from the parent class BaseDao.

I only added 2 scenarios: adding a product and deleting a product. In general, for practical consolidation of the article material, you can independently describe the method for editing products.

Let's immediately analyze the logic of deleting products from the database, since it is short.

When you click the "Delete product" button, a list of all products in our store appears. This list has a built-in "Delete" button, which stores a unique product identifier. In a real project, you could add sorting by categories, an internal search engine and other functions, but I decided not to complicate the task.

Loading of goods looks like this:

@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('Item deletion mode')
    all_products = await ProductDao.find_all(session=session_without_commit)

    await call.message.edit_text(
        text=f"Currently in the database {len(all_products)} products. To delete, click the button below"
    )
    for product_data in all_products:
        file_id = product_data.file_id
        file_text = "📦 Product with file" if file_id else "📄 Product without file"

        product_text = (f'🛒 Product Description:\n\n'
                        f'🔹 <b>Product name:</b> <b>{product_data.name}</b>\n'
                        f'🔹 <b>Description:</b>\n\n<b>{product_data.description}</b>\n\n'
                        f'🔹 <b>Price:</b> <b>{product_data.price} $</b>\n'
                        f'🔹 <b>Description (closed):</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))

Here's a little explanation. In the project's logic, "digital goods" means both simple text content, such as a link to download a file, and a specific file, such as a manual.

In this project, files are not stored physically, in the bot's folder, but on Telegram servers. This ensures high access speed and ease of use. In the context of the database, a file is a unique identifier that has been saved. To send a file, simply call the appropriate method.

await call.message.answer_document(document=file_id)

In the future, both when managing products from the admin panel and after users purchase files, we will send text content and documents in different ways. This is clearly demonstrated by the function described earlier.

Next, after clicking on the "Delete" button, the following function is launched:

@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"Product with ID {product_id} removed!", show_alert=True)
    await call.message.delete()

Here, notice that we are calling:

session_with_commit: AsyncSession

Since deleting information from the database implies committing.

This project does not provide a restriction mechanism. In real applications, the bot should ask: "Are you sure you want to delete?" - and only after the user confirms, delete the data. I decided not to waste time developing such functionality.

Describing the logic of adding products

In the admin panel, the most important and at the same time the most complex logic is built around adding a product to the database.

Here we will use the Finite State Machine technology (finite state machines) or, in other words, the scenario of "polling" the administrator when adding a product.

We have already prepared a class for FSM and now all that remains is to integrate it.

I want to emphasize that in this article I did not use the popular add-on to aiogram - Aiogram Dialog. Instead, I implemented everything through pure FSM Aigram 3.

A vote awaits you at the end of the article. If you want me to go through the Aiogram Dialog library in detail using a practical example, take part in the voting.

I would also like to draw your attention to another important point. Currently, we use MemoryStorage as a storage. Keep in mind that in this case, the information obtained during your survey will be stored in the bot's memory. If the bot reboots, it will "forget" everything you talked about.

Of course, this is not acceptable in production projects and you need to use more reliable storage, in particular RedisStorage.

Let's write a universal function that will interrupt the state machine scenario at any stage, wherever it is called:

@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('Cancel the Add Item Script')
    await call.message.delete()
    await call.message.answer(
        text="Cancel adding product.",
        reply_markup=admin_kb_back()
    )

Place this function somewhere at the beginning of the admin.py file.

It is important to understand that now all functions have state=[*] set by default. If you wrote on aiogram 2, you know that earlier you had to manually call this universal state, now it is by default.

In addition, in Aiogram 3, the handler for the ANY content type is now set by default. Therefore, do not forget to explicitly indicate which type of content should be processed through F-filters, for example.

Next, the administrator survey script begins and to save time, I will simply provide the entire survey code here, since there will be nothing remarkable or new from what was previously considered. Here is the code:

Hidden text

@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('The script for adding a product has been launched.')
    await call.message.delete()
    msg = await call.message.answer(text="First, please specify the name of the product: ", 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="Now give a short description of the product: ", 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="Now select the product category: ", 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('Product category successfully selected.')
    msg = await call.message.edit_text(text="Enter the price of the product: ", 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="Send a file (document) if required or click on 'NO FILE' if no file is required",
            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="Error! You must enter a numeric value for the price.")
        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('File not selected.')
    msg = await call.message.edit_text(
        text="Now send the content that will be displayed after purchasing the product inside the card",
        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="Now send the content that will be displayed after purchasing the product inside the card",
        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 = "📦 Product with file" if file_id else "📄 Product without file"

    product_text = (f'🛒 Check if everything is correct:\n\n'
                    f'🔹 <b>Product name:</b> <b>{product_data["name"]}</b>\n'
                    f'🔹 <b>Description:</b>\n\n<b>{product_data["description"]}</b>\n\n'
                    f'🔹 <b>Price:</b> <b>{product_data["price"]} $</b>\n'
                    f'🔹 <b>Description (closed):</b>\n\n<b>{product_data["hidden_content"]}</b>\n\n'
                    f'🔹 <b>Category:</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('I'm starting to save the file.!')
    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="The product has been successfully added to the database!", reply_markup=admin_kb())

If you have any questions, write here in the comments or in the community "Easy way to Python". There we have a friendly atmosphere and almost 2000 like-minded people.

At the end of the article, I will present a video demo of the working bot from both the user and admin parts, so if you are interested in seeing how the bot works in screen recording mode, scroll down.

At this point, our admin panel is ready, which means that we can proceed to the user part.

Writing the user part of the bot

File structure of the user part

We will work with the bot/user folder, in which we will create the following structure:

├── user/                
│   ├── __init__.py           # Batch file for easy imports
│   ├── user_router.py        # A file in which we describe general methods for the user: profile, viewing purchased goods, the “About Us” block
│   ├──catalog_router.py      # A file in which we describe methods for interacting with the catalog: viewing products, catalog page, purchasing products
│   ├── kbs.py                # Description of user keyboards
│   └── schemas.py            # Pydantic schemas for working with data

As you can see, we implement two routers for the user.

Let's start with the auxiliary files.

User keyboards (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="👤 My purchases", callback_data="my_profile")
    kb.button(text="🛍 Catalog", callback_data="catalog")
    kb.button(text="ℹ️ About the store", callback_data="about")
    kb.button(text="🌟 Support the author 🌟", url='https://t.me/tribute/app?startapp=deLN')
    if user_id in settings.ADMIN_IDS:
        kb.button(text="⚙️ Admin panel", 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="🏠 To the main page", callback_data="home")
    kb.adjust(2)
    return kb.as_markup()


def purchases_kb() -> InlineKeyboardMarkup:
    kb = InlineKeyboardBuilder()
    kb.button(text="🗑 View purchases", callback_data="purchases")
    kb.button(text="🏠 To the main page", callback_data="home")
    kb.adjust(1)
    return kb.as_markup()


def product_kb(product_id, price) -> InlineKeyboardMarkup:
    kb = InlineKeyboardBuilder()
    kb.button(text="💸 Buy", callback_data=f"buy_{product_id}_{price}")
    kb.button(text="🛍 Back", callback_data="catalog")
    kb.button(text="🏠 To the main page", callback_data="home")
    kb.adjust(2)
    return kb.as_markup()


def get_product_buy_kb(price) -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup(inline_keyboard=[
        [InlineKeyboardButton(text=f'Pay {price}$', pay=True)],
        [InlineKeyboardButton(text='Cancel', callback_data='home')]
    ])

Of all the keyboards, the only one that stands out here is the one with the “Pay” button.

def get_product_buy_kb(price) -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup(inline_keyboard=[
        [InlineKeyboardButton(text=f'Pay {price}$', pay=True)],
        [InlineKeyboardButton(text='Cancel', callback_data='home')]
    ])

I intentionally left it like this to show you that the classic approach of describing keyboards using InlineKeyboardMarkup is still supported in aiogram 3, although Builder is increasingly used.

Also, in this keyboard we use an unusual and perhaps unfamiliar button:

InlineKeyboardButton(text=f'Pay {price}$', pay=True)

It is important to understand that there can only be one payment button in one keyboard.

Otherwise, if you are familiar with the Aiogram 3 base, then the general logic of this code should be clear to you.

Pydantic Schemes (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="Telegram user ID")
    payment_id: str = Field(..., max_length=255, description="Unique payment ID")
    price: int = Field(..., description="Payment amount in usd")
    product_id: int = Field(..., description="Product ID")

This set includes both filtering schemes such as TelegramIDModel, ProductIDModel and ProductCategoryIDModel, and schemes that will help us check the data before storing it in tables. These include:

• PaymentData — for storing payment information;

• UserModel — for adding users to the database.

Now let's move on to describing the logic of the purchase process. After all, without a purchase, there would be no point in going to your personal account and the purchase section, right?

Working with the catalog router for the user (catalog_router.py)

We'll import and assign the router:

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()

Now let's describe the first page that the user will see when entering the catalog section.

@catalog_router.callback_query(F.data == "catalog")
async def page_catalog(call: CallbackQuery, session_without_commit: AsyncSession):
    await call.answer("Loading catalog...")
    catalog_data = await CategoryDao.find_all(session=session_without_commit)

    await call.message.edit_text(
        text="Select a product category:",
        reply_markup=catalog_kb(catalog_data)
    )

The only thing that deserves attention is the format of the keyboard formation with the catalog:

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="🏠 To the main page", callback_data="home")
    kb.adjust(2)
    return kb.as_markup()

Here we process the category information we received on the Aiogram side using a special function.

Next, using a loop, we divide each category into 2 buttons in a row, adding the "🏠 Home" button at the end. As a result, we get the following page:

Everything will work dynamically, and if you add a new entry to the category table, the new button will be pulled up automatically even without rebooting the bot.

Now let's describe the page that the user will see when clicking on a particular category:

@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"In this category {count_products} product.")
        for product in products_category:
            product_text = (
                f"📦 <b>Product name:</b> {product.name}\n\n"
                f"💰 <b>Price:</b> {product.price} usd.\n\n"
                f"📝 <b>Description:</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("There are no products in this category.")

Here the bot can either report that there are no products in the category or provide information about products without hidden content. In a full-fledged project, a search engine or a page navigation system could be integrated here.

When there are no products in the category

When there are products in the category

Please note that when displaying text, the formatting that was set at the stage of adding a product is preserved. We implemented this functionality earlier, as part of the admin panel scenario.

The point is that we saved not just text, but formatted text:

await state.update_data(description=message.html_text)

In the database, the text looks like this:

That is, the bot automatically transforms regular text into text with HTML tags, and thanks to the fact that when initializing the bot object, we did this with the following line:

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

We have set the transformation of text with HTML tags into formatted text by default.

Now we proceed to the most interesting part — implementing payment in the bot.

Connecting payment in the bot

The trigger for launching the payment scenario in our telegram bot will be a click on the "Buy" button. After that, this function is called:

@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'Payment 👉 {price}$',
        description=f'Please complete the payment in the amount of {price}$, to open access to the selected product.',
        payload=f"{user_info.id}_{product_id}",
        provider_token=settings.PROVIDER_TOKEN,
        currency='usd',
        prices=[LabeledPrice(
            label=f'Payment {price}',
            amount=int(price) * 100
        )],
        reply_markup=get_product_buy_kb(price)
    )
    await call.message.delete()

And let's take a closer look at this point.

In the Aiogram 3 library, the await bot.send_invoice method is used to send an invoice to the user for payment via Telegram Payments. This functionality allows bots to work with payment systems so that users can pay for goods or services directly in the chat with the bot.

Detailed description of the await bot.send_invoice method

The send_invoice method is an asynchronous function that sends a message with a description of the product, its price, and a payment button. After clicking the button, the user is redirected to the Telegram payment interface, where they complete the transaction.

Main parameters of the method:

1. chat_id:

• The ID of the chat where the message will be sent.

• Usually, call.from_user.id is used to send an invoice to the requestor.

1. title:

• Invoice title.

• Briefly describes the product or service.

1. description:

• Invoice description.

• For example, you can specify purchase details or remind what exactly you are paying for.

1. payload:

• Unique order identifier.

• Used to transfer data that will help you process the transaction (for example, user and product ID).

1. provider_token:

• Payment provider token issued by Telegram when setting up the payment system.

1. currency:

• Currency code (for example, 'usd' for usd).

1. prices:

• List of LabeledPrice objects describing the cost of the product.

• label — Item name (for example, "Payment 500").

• amount — Amount in the smallest currency units (e.g. 500 usd = 50,000 kopecks).

1. reply_markup:

• Custom keyboard or inline buttons that will appear with the invoice (optional).

How it works:

1. The user clicks the "Buy" button, calling the handler.

2. The bot sends the invoice using send_invoice.

3. Telegram displays the invoice with the "Pay" button.

4. The user finishes entering payment details.

5. Telegram sends your bot a pre_checkout_query event to confirm the order. The bot must handle this event within 10 seconds.

6. If the bot confirms the request, the transaction is completed and Telegram sends a successful_payment event.

Processing pre_checkout_query

The pre_checkout_query event is needed to check the order before completing the payment. For example, you can make sure that the product is in stock or check its current price.

An example of the simplest processing:

@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 — Unique request identifier that you use for the response.

• ok=True — Confirms that you are ready to complete the payment.

• If ok=False, the transaction will be canceled.

You can add some logic here, for example:

• Are there enough items in stock?

• Is the item price the same as the current one?

Successful payment processing (successful_payment)

After the transaction is completed, Telegram sends a successful_payment event. After processing it, you can save the payment details in the database and grant the user access to the purchased item.

Example implementation:

@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)
    }
    # Adding purchase information to the database
    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))

    # We are generating a notification for administrators
    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 {user_info} purchased the product <b>{product_data.name}</b> (ID: {product_id}) "
                    f"for <b>{product_data.price} $</b>."
                )
            )
        except Exception as e:
            logger.error(f"Error sending notification to administrators: {e}")

    # Preparing text for the user
    file_text = "📦 <b>The product includes a file:</b>" if product_data.file_id else "📄 <b>The product does not include files:</b>"
    product_text = (
        f"🎉 <b>Thank you for your purchase!</b>\n\n"
        f"🛒 <b>Information about your product:</b>\n"
        f"━━━━━━━━━━━━━━━━━━\n"
        f"🔹 <b>Name:</b> <b>{product_data.name}</b>\n"
        f"🔹 <b>Description:</b>\n<i>{product_data.description}</i>\n"
        f"🔹 <b>Price:</b> <b>{product_data.price} $</b>\n"
        f"🔹 <b>Hidden content:</b>\n<i>{product_data.hidden_content}</i>\n"
        f"━━━━━━━━━━━━━━━━━━\n"
        f"{file_text}\n\n"
        f"ℹ️ <b>You can find information about all your purchases in your personal profile.</b>"
    )

    # Sending product information to the user
    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)
        )

Summary

1. The send_invoice method sends an invoice to the user.

2. The pre_checkout_query handler confirms that the payment is ready to be completed.

3. After the successful_payment event, the bot saves the transaction data and provides the product to the user.

Here's how this implementation looks in the bot:

This is what issuing an invoice looks like.

Please note: as an administrator, I received a notification from the bot that a specific user made a purchase indicating the product and amount.

The payment block is already ready to work. To start accepting real payments, you will need to get a combat token through the Yukassa Telegram bot, installing it instead of the test token.

Let's describe the general user logic (user_router.py)

Let's perform imports and assign a router:

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()

Let's describe the function that will be executed when the user first logs into the bot and enters the /start command:

@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"👋 Hello, {message.from_user.full_name}! Select the required action",
            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>Thank you for registering!</b>. Now select the required action.",
                         reply_markup=main_user_kb(user_id))

Here we check whether the user is in the database or not. If he is, we register him, otherwise we just send him a message. In the demo, which will be presented at the end of the article, I will demonstrate working with a new user.

The following describes the function common with the admin panel for returning to the main menu:

@user_router.callback_query(F.data == "home")
async def page_home(call: CallbackQuery):
    await call.answer("Home page")
    return await call.message.answer(
        f"👋 Hello, {call.from_user.full_name}! Select the required action",
        reply_markup=main_user_kb(call.from_user.id)
    )

Here is a simple function with the "About Us" logic:

@user_router.callback_query(F.data == "about")
async def page_about(call: CallbackQuery):
    await call.answer("About the store")
    await call.message.answer(
        text=(
            "🎓 Welcome to our educational store!\n\n"
            "🚀 This bot was created as a demo project for an article on Habr.\n\n"
            "👨‍💻 Author: Yakovenko Alexey\n\n"
            "🛍️ Here you can study the principles of the telegram store, "
            "get acquainted with the functionality and mechanisms of interaction with the user.\n\n"
            "📚 This project is a great way to dive into the world of bot development "
            " and e-commerce in Telegram.\n\n"
            "💡 Explore, learn and be inspired!\n\n"
            "Data for test payment:\n\n"
            «Card: 1111 1111 1111 1026\n"
            "Sell ​​by: 12/26\n"
            "CVC-code: 000\n"
        ),
        reply_markup=call.message.reply_markup
    )

Next, we will describe the function that will be called when the user clicks on the “My purchases” button.

@user_router.callback_query(F.data == "my_profile")
async def page_about(call: CallbackQuery, session_without_commit: AsyncSession):
    await call.answer("Profile")

    # We get statistics of user purchases
    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)

    # We form a message depending on the availability of purchases
    if total_purchases == 0:
        await call.message.answer(
            text="🔍 <b>You have no purchases yet.</b>\n\n"
                 "Open the catalog and choose something interesting!",
            reply_markup=main_user_kb(call.from_user.id)
        )
    else:
        text = (
            f"🛍 <b>Your profile:</b>\n\n"
            f"Number of purchases: <b>{total_purchases}</b>\n"
            f"Total amount: <b>{total_amount}$</b>\n\n"
            "Would you like to view details of your purchases?"
        )
        await call.message.answer(
            text=text,
            reply_markup=purchases_kb()
        )

In short, here we check whether the user has made any purchases and, depending on this, we send one or another message with one or another keyboard.

It remains to describe the logic that will be called when clicking on "View purchases".

@user_router.callback_query(F.data == "purchases")
async def page_user_purchases(call: CallbackQuery, session_without_commit: AsyncSession):
    await call.answer("My purchases")

    # Getting the user's shopping list
    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>You have no purchases yet.</b>\n\n"
                 f"Open the catalog and choose something interesting!",
            reply_markup=main_user_kb(call.from_user.id)
        )
        return

    # For each purchase we send information
    for purchase in purchases:
        product = purchase.product
        file_text = "📦 <b>The product includes a file:</b>" if product.file_id else "📄 <b>The product does not include files:</b>"

        product_text = (
            f"🛒 <b>Information about your product:</b>\n"
            f"━━━━━━━━━━━━━━━━━━\n"
            f"🔹 <b>Name:</b> <i>{product.name}</i>\n"
            f"🔹 <b>Description:</b>\n<i>{product.description}</i>\n"
            f"🔹 <b>Price:</b> <b>{product.price} $</b>\n"
            f"🔹 <b>Hidden content:</b>\n<i>{product.hidden_content}</i>\n"
            f"━━━━━━━━━━━━━━━━━━\n"
            f"{file_text}\n"
        )

        if product.file_id:
            # We send a file with text
            await call.message.answer_document(
                document=product.file_id,
                caption=product_text,
            )
        else:
            # We send only text
            await call.message.answer(
                text=product_text,
            )

    await call.message.answer(
        text="🙏 Thank you for your trust!",
        reply_markup=main_user_kb(call.from_user.id)
    )

Now all that remains is to configure the main.py file and launch the bot.

Configuring main.py and launching the bot

In order for your bot to start working, you need to configure the main project file — bot/main.py. This file is responsible for registering routers, middleware, and functions for starting and stopping the bot. This is also where the command menu is configured and the bot is launched in long polling mode.

In real projects, it is better to use the web hook approach. In my profile on Habr you will find about five publications where I showed how to launch them via web hooks using FastApi and Aiohttp using different Telegram bots as an example.

Full code of the main.py file:

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

# Function that will customize the command menu (default for all users)
async def set_commands():
    commands = [BotCommand(command='start', description='Start')]
    await bot.set_my_commands(commands, BotCommandScopeDefault())

# The function that will be executed when the bot starts
async def start_bot():
    await set_commands()
    for admin_id in admins:
        try:
            await bot.send_message(admin_id, f'Im launched🥳.')
        except:
            pass
    logger.info("The bot has been launched successfully.")

# A function that will be executed when the bot has finished its work
async def stop_bot():
    try:
        for admin_id in admins:
            await bot.send_message(admin_id, f'Bot stopped. Why?😔')
    except:
        pass
    logger.error("Bot stopped!")

async def main():
    # Middleware Registration
    dp.update.middleware.register(DatabaseMiddlewareWithoutCommit())
    dp.update.middleware.register(DatabaseMiddlewareWithCommit())

    # Registering routers
    dp.include_router(catalog_router)
    dp.include_router(user_router)
    dp.include_router(admin_router)

    # Registering functions
    dp.startup.register(start_bot)
    dp.shutdown.register(stop_bot)

    # Launching the bot in long polling mode
    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())

Let's look at the key parts of the code:

1. Setting up the command menu

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

The function sets commands for the bot. In this case, the command /start is added with the description "Start".

2. Actions when starting the bot

async def start_bot():
    await set_commands()
    for admin_id in admins:
        try:
            await bot.send_message(admin_id, f'I'm launched🥳.')
        except:
            pass
    logger.info("The bot has been launched successfully.")

When the bot starts:

• Commands are configured using set_commands.

• Notification is sent to administrators.

• A message about the launch is recorded in the logs.

3. Actions when the bot stops

async def stop_bot():
    try:
        for admin_id in admins:
            await bot.send_message(admin_id, 'Bot stopped. Why?😔')
    except:
        pass
    logger.error("Bot stopped!")

When the bot stops:

• A notification is sent to administrators about the completion of work.

• A message about the stop is recorded in the logs.

4. Registration of middleware and routers

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)

• Middleware are intermediate functions that handle requests and responses. Middleware for working with the database is registered here.

• Routers are routers that group command and event handlers.

5. Launching the bot

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 — clears webhooks and removes pending updates. This entry in Aiogram 3 will work equally correctly both when using polling and when using webhooks.

• start_polling — starts the bot in long polling mode. The bot starts receiving updates from Telegram servers.

Now your bot is ready to work! It remains to run the main.py file and test its functionality.

To start the bot in the console, from the root of the project, enter the following command:

python -m bot.main

Below you can see a video showing the bot's operation process from both the user and administrative sides.

The last step remains to be able to confidently say that the project is ready - deploying the bot to the Amverum Cloud service.

Preparing and deploying the bot in the Amverum Cloud cloud service

Before launching the bot in the Amverum Cloud environment, it is necessary to carry out a number of preparatory activities. The key element of this process is the creation of the amvera.yml configuration file in the root directory of the project. This file contains important instructions that allow Amverum Cloud to correctly deploy and launch our bot.

The structure of the amvera.yml file

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

This configuration file provides Amverum with the following information:

• The project is written in Python version 3.12

• Dependencies should be installed from the requirements.txt file using pip

• Specifies the location of the database

• Specifies the command to run the bot

Pay special attention to the 'run' block

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

The persistenceMount: /data line is critical. It specifies that important files, including the SQLite database, should be mounted in the /data folder. This ensures that the data is not lost during a project rebuild, which may be necessary when updating the code.

After deployment, it is extremely important to verify that the database is indeed in the /data folder. This ensures that data is not lost during updates and rebuilds of the project.

Running the bot

The command python3 -m bot.main is used to run the bot. It is important to note that, unlike Windows, where the python command can be used, in a UNIX (Linux) environment, python3 must be used. This ensures that the bot runs correctly in the Amverum Cloud environment, which is based on a UNIX system.

Deployment process in Amverum Cloud

1. Registration: If you do not have an account yet, register in Amverum Cloud. New users receive a welcome bonus of 1$ to their account.

2. Creating a project: In the control panel, go to the projects section and click "Create a project".

3. Configuring a project: Assign a name to the project and select a suitable tariff plan. For educational purposes, the "Starter" tariff will be the best choice.

4. Uploading files: Select the method for uploading application files. Beginners are recommended to use the Amverum interface, more experienced users - GIT commands.

5. Check settings: On the next screen, make sure all settings have loaded correctly and click "Finish".

6. Waiting for deployment: The deployment process will take 3-4 minutes. Once completed, you will receive a notification from the bot about its successful launch, and an indicator of the active status of the application will appear in the Amverum interface.

If you are interested, you can click the finished bot here.

Let me remind you that the full source code of today's project and exclusive content that I do not publish on Habr can be found in my free telegram channel "Easy Path to Python".

Conclusion

Today we have done serious work, and understanding the code that we have discussed will open up new horizons for you in developing complex and functional Telegram bots with an emphasis on Internet marketing.

This experience will also lay the foundation for organizing projects, integrating databases and applying a microservice approach, which is useful not only in FastAPI, but also in the world of Telegram bots.

What we have learned:

• Developing an admin panel for a Telegram bot: We have created a convenient interface for managing the bot.

• Integrating SQLAlchemy 2: We have learned how to use this powerful ORM to work with data in Telegram bots.

• Working with the state machine: We mastered state management in the bot, which allows us to create interactive scenarios.

• Payment system integration: We went through all the steps - from issuing an invoice to processing a successful payment and sending notifications to the bot.

I understand that there is a lot of material, but I tried to make it as accessible as possible, even if it takes you some time to digest it all.

If this article was useful to you - let me know with a like or a comment! Sometimes the work I spent several days on can go unnoticed, and at such moments I think about the fact that no one needs all this.

Don't forget to also take part in the voting under this article. I'm interested in knowing if you are interested in the topic of integrating payment by stars and direct payment bypassing BotFather with webhooks.

That's all from me. See you soon!