DEV Community: Amverum Cloud

Asynchronous SQLAlchemy 2: A simple step-by-step guide to configuration, models, relationships, and migrations using Alembic

Amverum Cloud — Tue, 03 Jun 2025 06:45:58 +0000

Finally, it's time to do what I've been planning for a long time - create a detailed guide to working with the asynchronous version of SQLAlchemy 2.0 in the ORM style. This series of articles will cover absolutely all aspects: from models and relationships between them to migrations and methods of interacting with data in the database.

What's planned?

I'm going to write several articles that will be balanced between the necessary "boring" theory and practical examples to help everyone who is already familiar with Python master this "magical" framework. And believe me, you will soon understand that alchemy in the name of the framework was not chosen by chance.

Today's Topic

First, let's understand what SQLAlchemy is and why every developer working with relational databases (such as SQLite, PostgreSQL, MySQL, etc.) should know about it. After that, the setup. We'll be working with PostgreSQL, but don't worry: the code we'll write is universal for all relational databases. We'll start with the basic setup of SQLAlchemy for asynchronous interaction, and then move on to creating tables in a modern declarative style.

Table Models

Today, we'll learn how to describe table models and discuss different types of columns - from common ones (such as Integer, String, Text) to specific ones (Array, JSON).

Table models in SQLAlchemy are a kind of Python classes that directly reflect the structure of tables in the database. Each model describes a specific table, its columns and their data types. Models allow you to work with database records as with regular objects, which greatly simplifies interaction with the database.

Today we will look at how to establish relationships at the database level using foreign keys (ForeignKey), as well as how to use more advanced relationships in SQLAlchemy, which greatly simplify working with data thanks to the use of relationship. Let's start with the basic types of relationships: "One to One" (1:1), "One to Many" (1:N) and "Many to One" (N:1).

The only thing we will leave aside for now is "Many to Many" (N:M) relationships and indexes. We will talk about these topics later, in the following articles.

After we have a foundation in the form of table models, we will create the tables themselves. For this, we will use the Alembic library, which allows you to easily manage database migrations and integrates perfectly with SQLAlchemy. Thanks to this tool, our application will evolve along with the database, which is especially important when working in a team.

A bit of theory

For those who are just getting acquainted with SQLAlchemy, it is important to understand that this is not just a framework, but a full-fledged tool for working with relational (tabular) databases. It supports two styles: core and ORM.

Core and ORM

Core is a low-level approach that allows you to execute queries using SQL expressions, providing full control over the process. This style is suitable for those who want to get as close as possible to standard SQL or have special performance requirements.
ORM (Object-Relational Mapping) is a style in which the framework maps database tables to Python classes. With ORM, you work with objects, not with SQL strings. This is the style we will focus on, as it is more convenient, universal and popular among developers.

ORM Core Components

Working with SQLAlchemy in an ORM style involves several key concepts:

Table Models are Python classes that represent database tables. These classes contain information about the structure of tables, such as columns, data types, and relationships between tables.
Sessions are objects through which interaction with the database is carried out. They allow you to execute queries and commit changes. A session is opened at the beginning of work with the database and closed at the end, providing communication with the database for the duration of one “session”.
Session Factory is a template for creating sessions. It is used to manage the connection to the database and create new sessions as needed.

Why should every Python developer know SQLAlchemy?

SQLAlchemy simplifies working with databases by turning them into intuitive Python objects. This makes your code cleaner and more readable, since you are writing Python, not SQL. In addition, it allows you to work with different databases with almost no code changes — a great feature for those working in teams that require different types of databases. SQLAlchemy also offers powerful tools for managing relationships and migrations, making it a versatile choice for developing large projects.

Relationships between tables

As mentioned earlier, SQLAlchemy allows you to establish relationships between tables through foreign keys. Let's look at the main types of relationships:

"One to One" (1:1) — used when each record in one table should correspond to only one record in another table. For example, a user profile can be related to a user account in a one-to-one relationship.
"One to Many" (1:N) — with this relationship, one record in one table can correspond to multiple records in another table. For example, one user can have multiple blog posts.
"Many to One" (N:1) — a reverse "one to many" relationship. In this case, multiple records from one table can refer to a single record in another table, for example, multiple comments can be linked to a single post.

Such relationships allow you to build complex data structures and simplify access to related data.

Let's prepare the database for work

We have several options for preparing the database for work using PostgreSQL as an example. Let's look at them:

Installing PGAdmin on a computer

Go to the official PGAdmin website and download the latest version.
Install the program, do not forget to remember the password that you create during installation.
After installation, the PostgreSQL database will work locally on your computer.

Use the following parameters to connect:

Host: localhost
Port: 5432
Database: postgres
User: postgres

This method provides access to the database on your computer only.

Deploying PostgreSQL via a Docker container

Make sure you have Docker and Docker Compose installed on your computer.

Follow the instructions in this article to set up a database via Docker-compose.

When running locally, access will only be from your computer, and when running on a VPS server, you can configure global access.

The connection parameters will be the same as those you specified when configuring the Docker Compose file.

Deploying PostgreSQL using a service

This approach is convenient because all issues related to setting up, optimizing, and running the database are taken care of by the hosting service. One such service is Amverum Cloud.

You just need to register, get a bonus of $1 for registration and launch PostgreSQL with external access in a couple of minutes. Below I will tell you how to do this, since I choose this method myself.

Raising PostgreSQL with external access on Amverum Cloud

Register in Amverum Cloud, if you have not done so yet
Go to the projects section
Click on "Create a project". Next, give the project a name, select the "Database" service type and select a tariff. "Start" is suitable for educational purposes.

The name does not have to match the database name.

On the new screen, select the database type, give the database a name, come up with a username and password, connect the superuser and click "finish". We also set the cluster size for the DBMS to 1. This is enough for educational purposes.

Open external access to the database. To do this, go to the created project. There, go to the "Settings" tab. On the screen that opens, click on "Add free domain name". Select the connection type "POSTGRES". Then click on "Apply".

Now we will have a free domain, which will become a host for connection.

Here are the data for connecting to the database:

HOST is the address you got in the last step.
USER is either the user you created or the postgres superuser.
PASSWORD is the password you created yourself.
DATABASE is the database you created.

If you suddenly forget some parameters, you can always open the "Configuration" tab in your project and view the necessary information.

Testing the connection to the database

Regardless of which type of database recovery you have chosen, before we start working with SQLAlchemy, let's perform a test connection to the database.

For this, I suggest installing the free and convenient DBeaver Community program. It works universally with all tabular databases, including PostgreSQL.

Download DBeaver Community for your operating system
Launch
Click on "New connection" and select PostgreSQL from the list

Specify the connection details. I take the data received in Amverum Cloud

Connection established successfully!

Then click on the "Test connection" button. If you see the result as in the screenshot above, you can click "Done". The connection is established successfully.

Preparing the project

Now that PostgreSQL is ready to work, let's start setting up the project in the IDE. I choose PyCharm, but it can be Atom, VSCode and other editors.

Create a project in PyCharm and activate the virtual environment.
Create a .env file to store environment variables for connecting to the database:

Example:

DB_HOST=alchemsy-yakvenalex.waw.amverum.cloud
DB_PORT=5232
DB_NAME=adlchemydb
DB_USER=admin
DB_PASSWORD=dsakjjASSDkk

Create a requirements.txt file and add the necessary libraries:

alembic==1.13.3
sqlalchemy==2.0.35
pydantic==2.9.2
pydantic-settings==2.5.2
asyncpg==0.29.0

Install dependencies:

pip install -r requirements.txt

Brief description of modules

Alembic: A tool for managing and migrating database schema. Used to make changes to the database structure (adding, deleting, changing tables and columns).
SQLAlchemy: The main library for working with databases in Python, providing ORM (object-relational mapping) and tools for executing SQL queries.
Pydantic: A library for data validation and data type management in Python. Pydantic allows you to conveniently define data schemas and check them for compliance with types. In the future, we will use this library to transform data from SQLAlchemy into familiar objects, such as dictionaries.
Pydantic-settings: An add-on to Pydantic that allows you to easily manage application configuration settings using Pydantic for validation and typing. We will use it to work with environment variables that will allow us to connect to a PostgreSQL database.
Asyncpg: Asynchronous engine for working with PostgreSQL via SQLAlchemy.

Creating a database

Now it's time to decide what tables we will create and how to build the application logic around them. As a visual example, let's take a simplified version of any blog.

We will have users with a minimal set of data in the main table, which will be related to other tables:

Profile table - 1:1 relationship. Each user can have only one profile.
Posts table - 1:N relationship. One user can have many posts, but each post can only be written by one user.
Comments table - 1:N relationship. One user can have many comments, and each comment can only be written by one user.

This set of models will allow us to demonstrate how to describe tables and establish relationships between them for effective work with data.

This should give you a good understanding of how SQLAlchemy works and how to build models for different data types and relationships. Time to put it into code!

Prepare the configuration file

In the root of the project, I will create a file config.py. Its purpose is to output environment variables to the project and generate a link to connect to the PostgreSQL database. For these tasks, I will use pydantic-settings.

Here is a complete code example:

import os
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    DB_USER: str
    DB_PASSWORD: str
    DB_HOST: str
    DB_PORT: int
    DB_NAME: str

    # DATABASE_SQLITE = 'sqlite+aiosqlite:///data/db.sqlite3'
    model_config = SettingsConfigDict(
        env_file=os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
    )

    def get_db_url(self):
        return (f"postgresql+asyncpg://{self.DB_USER}:{self.DB_PASSWORD}@"
                f"{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}")


settings = Settings()

Here we created a Settings class, placing all the variables from the .env file into it. In addition, we described a method that will allow us to generate a link for an asynchronous connection to the PostgreSQL database via SQLAlchemy.

Next, I assigned the settings variable as an object of the Settings class. Now we have the ability to access the methods and variables we need via a dot. An example is below:

print("DB URL =>", settings.get_db_url())
print("DB HOST =>", settings.DB_HOST)

In the class I left an example link for connecting to SQLite. It uses the aiosqlite engine for asynchronous work.

Database configuration file

To work with a database in SQLAlchemy, a separate file is usually created, where the main settings are written. We will call it database.py. Here is the minimum configuration:

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.ext.asyncio import AsyncAttrs, async_sessionmaker, create_async_engine
from config import settings


DATABASE_URL = settings.get_db_url()

# We create an asynchronous engine for working with a database
engine = create_async_engine(url=DATABASE_URL)
# Create a session factory to interact with the database
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)


# Base class for all models
class Base(AsyncAttrs, DeclarativeBase):
    __abstract__ = True  # The class is abstract so that you don't have to create a separate table for it

Configuration Description

DeclarativeBase: The base class for all models, from which all tables (table models) will inherit. We will use this feature of the class many times.
AsyncAttrs: Allows you to create asynchronous models, which improves performance when working with asynchronous operations.
create_async_engine: A function that creates an asynchronous engine for connecting to the database at the provided URL.
async_sessionmaker: A session factory for asynchronous interaction with the database. Sessions are used to perform queries and transactions.

How it works

DATABASE_URL: Generated using the get_db_url method from the config.py configuration file. Contains all the information necessary to connect to the database.
engine: Asynchronous engine required to perform database operations.
async_session_maker: Session factory that allows creating sessions for interacting with the database, managing transactions and executing queries.
Base: Abstract base class for all models, from which all tables will be inherited. It will not create a separate table in the database, but will provide basic functionality for all other models.

This minimal set of settings provides a basic configuration for working with the database in SQLAlchemy. In the future, in this and the following articles, we will expand this file, adding new methods and settings for more complex tasks.

Let's expand the settings code

In order to better understand how this works, we will immediately add to our base class a description of three universal columns that will be substituted for each table created.

We will describe these columns only once, and they will be created in each table. In addition, let's create a universal method that will name our tables according to the following scheme: "table model name" + "s" with translation to lower case.

For example, if we had a model name User, then the table name will be users.

Here is the full modified code:

from datetime import datetime
from sqlalchemy import Integer, func
from sqlalchemy.orm import DeclarativeBase, declared_attr, Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncAttrs, async_sessionmaker, create_async_engine

from config import settings

DATABASE_URL = settings.get_db_url()

# We create an asynchronous engine for working with a database
engine = create_async_engine(url=DATABASE_URL)
# Create a session factory to interact with the database
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)

# Base class for all models
class Base(AsyncAttrs, DeclarativeBase):
    __abstract__ = True  # The class is abstract so that you don't have to create a separate table for it

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(server_default=func.now(), onupdate=func.now())

    @declared_attr.directive
    def __tablename__(cls) -> str:
        return cls.__name__.lower() + 's'

The Base base class defines three columns (id, created_at, updated_at) that will be added to all models inheriting from Base:

id: Primary key, which is automatically incremented.
created_at: Date and time of record creation, set automatically.
updated_at: Date and time of the last update of the record, automatically updated with each change.

Let's leave this file as is for now, but we'll get back to it soon.

You've probably noticed an unfamiliar syntax for describing table columns, namely Mapped and mapped_column. Let's figure out what it is and how to work with it.

Mapped

Mapped is a modern way to annotate data types for columns in SQLAlchemy models. It allows you to more clearly indicate that a variable represents a column of a table in a database, making the code more readable and understandable.

Example:

id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

Here we specify that id is an int column that will be used as a primary key and will be automatically incremented.

mapped_column

mapped_column is a function that is used to create columns in SQLAlchemy models. It takes as arguments the data type of the column and additional parameters such as primary_key, nullable, default, etc.

Example:

created_at: Mapped[datetime] = mapped_column(server_default=func.now())

Here we create a created_at column that will be automatically populated with the current date and time when a record is created.

How to use it

Using Mapped and mapped_column we can describe models more concisely and clearly.

Example for the user model:

from sqlalchemy import Integer, String
from sqlalchemy.orm import Mapped, mapped_column


class User(Base):
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String, nullable=False)
    email: Mapped[str] = mapped_column(String, unique=True, nullable=False)

In this example:

id: An integer primary key that is automatically incremented.
name: A string that cannot be empty (nullable=False).
email: A unique string that also cannot be empty.

Maximum flexibility and minimum code

With Mapped and type annotations, we can greatly simplify the description of models.

Example:

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncAttrs, DeclarativeBase


class Base(AsyncAttrs, DeclarativeBase):
    __abstract__ = True


class User(Base):
    __tablename__ = 'users'

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str]
    surname: Mapped[str | None]
    email: Mapped[str] = mapped_column(unique=True)

What it means:

name: Mapped[str]: name is required (nullable=False by default), of type string. No need to explicitly specify mapped_column(String, nullable=False).
surname: Mapped[str | None]: surname is optional (nullable=True), since the data type indicates that it can be None. No need to explicitly specify mapped_column(String, nullable=True).

Combinations:

The email field shows how we can add additional parameters to mapped_column if needed.

Benefits

Code Cleanliness: Simplifies and makes code more readable.
Type Explicitness: Indicates the data type up front, which is useful for understanding the data structure.
Flexibility: Allows easy control over whether fields are required.

Full Example

Here is how a full user model can be described using these principles:

from sqlalchemy import String, Integer, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncAttrs, DeclarativeBase


class Base(AsyncAttrs, DeclarativeBase):
    __abstract__ = True


class User(Base):
    __tablename__ = 'users'

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str]
    surname: Mapped[str | None]
    email: Mapped[str] = mapped_column(unique=True)
    created_at: Mapped[DateTime] = mapped_column(server_default=func.now())
    updated_at: Mapped[DateTime] = mapped_column(server_default=func.now(), onupdate=func.now())

Later I will show how to simplify and shorten the code using Mapped.

In other words, if you need something "standard", use pure Mapped, and if you need a more detailed description, add mapped_column.

User model

Let's start describing the models. The first model we will describe is the user model. For now, we will not include SQLAlchemy relationships (relationship) in order not to complicate the process of writing code.

From the "unusual" we use foreign keys (ForeignKey), thereby laying the foundation for future relationships.

What is ForeignKey in simple terms

Simply put, ForeignKey is a way to link one table to another in the database. Imagine that you have two tables: users and posts. Each post needs an author, right?

This is where ForeignKey comes into play. In the posts table, we add a column that will store the user ID from the users table. This allows SQLAlchemy (and the database) to know who the author of each post is.

Code example:

from sqlalchemy import Integer, String, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.ext.asyncio import AsyncAttrs, DeclarativeBase


class Base(AsyncAttrs, DeclarativeBase):
    __abstract__ = True


class User(Base):
    __tablename__ = 'users'

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str]


class Post(Base):
    __tablename__ = 'posts'

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    title: Mapped[str]
    content: Mapped[Text]
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))  # Foreign key

This allows you to link each post to a specific user, which makes working with data much more convenient and structured. Further, in this bundle, relationship will be used - a dependency on the SQLAlchemy side, which we will also consider later.

Create a models.py file

Create a models.py file in the root of the project and describe the first model there:

from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column
from database import Base

class User(Base):
    username: Mapped[str] = mapped_column(unique=True)
    email: Mapped[str] = mapped_column(unique=True)
    password: Mapped[str]
    profile_id: Mapped[int | None] = mapped_column(ForeignKey('profiles.id'))

Note that we haven't specified a table name or assigned a primary key (user ID). We don't need to, because the Base class will automatically add these columns when creating the table and assign it the name users.

Optimizing code with annotations

We also noticed that the line Mapped[str] = mapped_column(unique=True) is repeated several times. To optimize this process, we can use annotations.

Annotations in SQLAlchemy allow you to move the description of columns to separate variables. This means that we can describe the columns once and then use them in models. This approach simplifies the code and makes it more readable.

To use annotations, you need to import the Annotated object from the typing module. I usually describe annotations in the database.py file. Here is an example:

from typing import Annotated
from sqlalchemy import String
from sqlalchemy.orm import mapped_column


uniq_str_an = Annotated[str, mapped_column(unique=True)]

Annotation Description

Annotated is a tool from the typing module that allows you to add metadata to data types. In this case, we use it to describe a column in SQLAlchemy.
str is a Python data type that specifies that the column will be a string.
mapped_column(unique=True) is a function that specifies that the column will be unique, meaning that two values in this column cannot be repeated.

An example of using annotation:

from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column
from database import Base, uniq_str_an


class User(Base):
    username: Mapped[uniq_str_an]
    email: Mapped[uniq_str_an]
    password: Mapped[str]
    profile_id: Mapped[int | None] = mapped_column(ForeignKey('profiles.id'))

The code is more readable now, and it's starting to make sense why the framework we're talking about is called SQLAlchemy, right?

We won't go into detail about data types, since if you've read this far, you probably have a basic understanding of Python and database data types.

User Profile Table

Now we'll describe the user profile model (table). We'll include a few extra data types here to reinforce the point.

I suggest adding the following fields:

First name: string, required field
Last name: string, optional field
Age: integer
Gender: male / female — let's make one of the two values (ENUM) work
Profession: one of the provided values (ENUM)
Interests: list (Array) of strings
Contacts: JSON (Python dictionary) with an arbitrary list of contacts.

Before describing the model itself, let's prepare a set of values for gender and professions. To do this, let's first import enum (regular enum, not from SQLAlchemy).

import enum

Enum from the enum module in Python is used to create enumerations, which are sets of named values. This allows you to define data types with a limited set of possible values.

Enumerations for gender:

class GenderEnum(str, enum.Enum):
    MALE = "male"
    FEMALE = "female"

Listings for professions:

class ProfessionEnum(str, enum.Enum):
    DEVELOPER = "developer"
    DESIGNER = "designer"
    MANAGER = "manager"
    TEACHER = "teacher"
    DOCTOR = "doctor"
    ENGINEER = "engineer"
    MARKETER = "marketer"
    WRITER = "writer"
    ARTIST = "artist"
    LAWYER = "lawyer"
    SCIENTIST = "scientist"
    NURSE = "nurse"
    UNEMPLOYED = "unemployed"

Now let's describe the model itself.

class Profile(Base):
    first_name: Mapped[str]
    last_name: Mapped[str | None]
    age: Mapped[int | None]
    gender: Mapped[GenderEnum]
    profession: Mapped[ProfessionEnum] = mapped_column(
        default=ProfessionEnum.DEVELOPER,
        server_default=text("'UNEMPLOYED'")
    )
    interests: Mapped[List[str] | None] = mapped_column(ARRAY(String))
    contacts: Mapped[dict | None] = mapped_column(JSON)

Unfortunately, the ARRAY and JSON data types cannot be described via pure Mapped yet, so I used mapped_column here. Everything else will work correctly.

Using default and server_default parameters

You probably noticed that we used the default and server_default parameters for the profession. Let's take a closer look at how they differ and when to use them.

Parameter default:

This parameter sets the default value at the application (SQLAlchemy) level.
This means that if a value for this field is not specified when creating an object in the code, the value specified in default will be used. For example, when creating an object of the User class, if a value for the profession field is not passed, SQLAlchemy will automatically substitute the default value specified in default.
Example: If we have an enumeration (ENUM) of professions, then the default value can be selected using a dot, for example: ProfessionEnum.DEVELOPER.

Parameter server_default:

This parameter sets the default value at the database level.
This means that if a value for this field is not specified when inserting a record into a table, the database itself will substitute the value specified in server_default. Unlike default, this value is applied if a record is added to a table directly, for example, via SQL queries, bypassing the application.
Important: To use this parameter with ENUM, you must pass the value as a text expression using the text method, which is imported from SQLAlchemy. The ENUM value is specified in quotes as text, for example: "WRITER", and not the value itself, such as ProfessionEnum.WRITER. This is necessary for the correct execution of the query on the database side.

Model for posts

Now we will describe a simple model for posts. I will prepare a model with values for the post publication status in advance.

class StatusPost(str, enum.Enum):
    PUBLISHED = "published"
    DELETED = "removed"
    UNDER_MODERATION = "under moderation"
    DRAFT = "draft"
    SCHEDULED = "delayed publication"

Now let's describe the model itself

class Post(Base):
    title: Mapped[str]
    content: Mapped[Text]
    main_photo_url: Mapped[str]
    photos_url: Mapped[List[str] | None] = mapped_column(ARRAY(String))
    status: Mapped[StatusPost] = mapped_column(default=StatusPost.PUBLISHED, server_default=text("'DRAFT'"))
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))

A fairly simple model that will contain all the required fields: title, text content (let's assume HTML-formatted text), link to the main photo, links to additional photos, publication status from the set, and user_id for the link.

The list description field has already been encountered earlier. Let's briefly outline it in the annotation.

array_or_none_an = Annotated[List[str] | None, mapped_column(ARRAY(String))]

In our current code, enumerations (ENUMS) take up a lot of space. Let's move them to a separate file called sql_enums.py.

import enum


class GenderEnum(str, enum.Enum):
    MALE = "male"
    FEMALE = "female"


class StatusPost(str, enum.Enum):
    PUBLISHED = "published"
    DELETED = "removed"
    UNDER_MODERATION = "under moderation"
    DRAFT = "draft"
    SCHEDULED = "delayed publication"


class ProfessionEnum(str, enum.Enum):
    DEVELOPER = "developer"
    DESIGNER = "designer"
    MANAGER = "manager"
    TEACHER = "teacher"
    DOCTOR = "doctor"
    ENGINEER = "engineer"
    MARKETER = "marketer"
    WRITER = "writer"
    ARTIST = "artist"
    LAWYER = "lawyer"
    SCIENTIST = "scientist"
    NURSE = "nurse"
    UNEMPLOYED = "unemployed"

Model with comments

In this model, we will need to associate a comment with both the author of the comment and the post to which this comment was left.

Also, as a demonstration, let's add here the is_publish column - a Boolean value in the format of whether the comment was published or not. Just to see how such fields are described.

And add a rating from 1 to 10, to consolidate the topic with ENUM.

Describe ENUM

class RatingEnum(int, enum.Enum):
    ONE = 1
    TWO = 2
    THREE = 3
    FOUR = 4
    FIVE = 5
    SIX = 6
    SEVEN = 7
    EIGHT = 8
    NINE = 9
    TEN = 10

Now let's describe the model itself:

class Comment(Base):
    content: Mapped[Text]
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    post_id: Mapped[int] = mapped_column(ForeignKey('posts.id'))
    is_published: Mapped[bool] = mapped_column(default=True, server_default=text("'false'"))
    rating: Mapped[RatingEnum] = mapped_column(default=RatingEnum.FIVE, server_default=text("'SEVEN'"))

If you have been reading this article carefully, by now you should have a full understanding of how to describe columns using the SQLAlchemy ORM in a modern style. Now we will move on to a more complex topic - describing dependencies between tables.

Defining dependencies

Here I will try to explain the topic of dependencies in as much detail as possible, as I know that beginners often have difficulties at this stage. My goal is to help you fully understand how to describe relationships in SQLAlchemy models.

One-to-One

1. Example of a relationship

We have two models: User and Profile. The relationship between them is organized according to the principle of One-to-One. This means that each user has only one profile, and each profile belongs to only one user.

2. Why One-to-One?

Typically, a one-to-one relationship is used to separate primary and secondary information. For example, user information (name, email) can be stored in the User table, and profile data (age, profession, interests) is stored in a separate Profile table. This avoids unnecessary width of the main table, which simplifies its maintenance and data management.

3. Technical implementation

In SQLAlchemy, the relationship technology allows you to set up and manage relationships between tables. In the case of a one-to-one relationship between the User and Profile tables, the setup looks like this:

Relationship in the User model:

profile: Mapped["Profile"] = relationship(
    "Profile",
    back_populates="user",
    uselist=False,  # Key parameter for one-to-one communication
    lazy="joined"  # Automatically loads profile when user requests
)

back_populates="user" — specifies the back attribute of the Profile model. This means that when accessing a profile, you can also get the associated user.
uselist=False — specifies that the relationship is not a list (by default, relationship assumes a one-to-many or many-to-one relationship). Here, one profile per user, so uselist=False specifies a one-to-one relationship.
lazy="joined" — specifies the data loading strategy. SQLAlchemy will perform a JOIN query and load the profile immediately when the user is requested.

Feedback in the Profile model:

user: Mapped["User"] = relationship(
    "User",
    back_populates="profile",
    uselist=False
)

back_populates="profile" — links this relationship to the profile attribute on the User model, allowing for bidirectional control. If you link a user and a profile, both attributes (user and profile) will be updated at the same time.
uselist=False — since each profile can only have one user, False indicates that this relationship does not support lists, but implies a single object.

How does it work?

When a User object is created with a Profile attached, SQLAlchemy automatically links them via the relationship definition. For example, if we get a user via an ORM select query, SQLAlchemy will perform a JOIN query and load the profile if joinedload is used for the lazy='joined' option. We will cover this in more detail in the next article.

One-to-Many and Many-to-One Relationships

1. Relationship Example

In our User model, the relationship with the Post model is one-to-many: one user can create many posts, but each post belongs to only one user. This relationship is also called many-to-one on the Post side, since multiple posts can refer to the same user.

Similarly, the Post model has a one-to-many relationship with Comment: each post can have multiple comments, but each comment refers to a specific post.

2. Why use One-to-Many and Many-to-One?

A one-to-many or many-to-one relationship is used when one object (such as a user) is related to many other objects (such as posts). This allows you to logically group data and avoid redundancy. In our case, each user can have multiple posts, and each post can have multiple comments.

3. Technical implementation

One-to-many and many-to-one relationships can be configured using relationship and ForeignKey.

Relationship between User and Post:

class User(Base):
    # User fields...

    posts: Mapped[list["Post"]] = relationship(
        "Post",
        back_populates="user",
        cascade="all, delete-orphan"  # Deletes posts when user is deleted
    )

class Post(Base):
    # Post fields...

    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))

    user: Mapped["User"] = relationship(
        "User",
        back_populates="posts"
    )

user_id: In Post, this is a foreign key that links each post to a specific user.
back_populates:

back_populates="user" in Post indicates a relationship with the posts field in User.
back_populates="posts" in User links with the user field in Post. These settings ensure a two-way relationship.

cascade="all, delete-orphan": This setting in the User model specifies that all posts associated with the user should be deleted if the user itself is deleted.

Relationship between Post and Comment:

class Post(Base):
    # Post fields...

    comments: Mapped[list["Comment"]] = relationship(
        "Comment",
        back_populates="post",
        cascade="all, delete-orphan"
    )

Here, thanks to relationship, you can associate objects with each other when you create them, and SQLAlchemy will automatically determine which records they belong to.

Try to associate comments and users yourself before you open the full code. I emphasize that understanding these concepts is very important for a general understanding of interaction with data.

Did you write it? Great, then you can open the spoiler with the full code.

Hidden text

from sqlalchemy import ForeignKey, JSON, text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from database import Base, uniq_str_an, array_or_none_an, content_an
from sql_enums import GenderEnum, ProfessionEnum, StatusPost, RatingEnum


class User(Base):
    username: Mapped[uniq_str_an]
    email: Mapped[uniq_str_an]
    password: Mapped[str]
    profile_id: Mapped[int | None] = mapped_column(ForeignKey('profiles.id'))

    # One-to-one connection with Profile
    profile: Mapped["Profile"] = relationship(
        "Profile",
        back_populates="user",
        uselist=False,  # Provides one-to-one communication
        lazy="joined"  # Automatically loads related data from Profile when a User is requested
    )

    # One-to-many relationship with Post
    posts: Mapped[list["Post"]] = relationship(
        "Post",
        back_populates="user",
        cascade="all, delete-orphan"  # When deleting a User, the associated Post is also deleted
    )

    # One-to-many relationship with Comment
    comments: Mapped[list["Comment"]] = relationship(
        "Comment",
        back_populates="user",
        cascade="all, delete-orphan"  # When deleting a User, the associated Comment is also deleted
    )


class Profile(Base):
    first_name: Mapped[str]
    last_name: Mapped[str | None]
    age: Mapped[int | None]
    gender: Mapped[GenderEnum]
    profession: Mapped[ProfessionEnum] = mapped_column(default=ProfessionEnum.DEVELOPER,
                                                       server_default=text("'UNEMPLOYED'"))
    interests: Mapped[array_or_none_an]
    contacts: Mapped[dict | None] = mapped_column(JSON)

    # One-to-one feedback with User
    user: Mapped["User"] = relationship(
        "User",
        back_populates="profile",
        uselist=False
    )


class Post(Base):
    title: Mapped[str]
    content: Mapped[content_an]
    main_photo_url: Mapped[str]
    photos_url: Mapped[array_or_none_an]
    status: Mapped[StatusPost] = mapped_column(
        default=StatusPost.PUBLISHED,
        server_default=text("'DRAFT'"),
        nullable=False
    )

    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))

    # Many-to-one relationship with User
    user: Mapped["User"] = relationship(
        "User",
        back_populates="posts"
    )

    # One-to-many relationship with Comment
    comments: Mapped[list["Comment"]] = relationship(
        "Comment",
        back_populates="post",
        cascade="all, delete-orphan"
    )


class Comment(Base):
    content: Mapped[content_an]
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    post_id: Mapped[int] = mapped_column(ForeignKey('posts.id'))
    is_published: Mapped[bool] = mapped_column(default=True, server_default=text("'false'"))
    rating: Mapped[RatingEnum] = mapped_column(default=RatingEnum.FIVE, server_default=text("'SEVEN'"))

    # Many-to-one relationship with User
    user: Mapped["User"] = relationship(
        "User",
        back_populates="comments"
    )

    # Many-to-one relationship with Post
    post: Mapped["Post"] = relationship(
        "Post",
        back_populates="comments"
    )

Real Table Creation

We have described the models of our future tables, but to turn them into real tables in the database, we need to perform the migration process. SQLAlchemy offers several ways to create and update tables based on models, and one of the most popular and powerful tools for this task is Alembic.

Alembic allows you to manage versions of the database schema, track changes and easily migrate them to different environments. This tool simplifies the creation and execution of migrations, keeping the database up to date as the models change.

Let's look at how exactly we can use Alembic to migrate our models to the database to simplify further work with data and maintain convenience and control over changes.

To start working with Alembic, we need to initialize it with support for asynchronous interaction with the database. This can be done with the following command:

alembic init -t async migration

After running this command, a migration directory and an alembic.ini file will appear in the root of the project. The migration directory will contain files for managing migrations, and alembic.ini is the Alembic configuration file that we will need to set up a connection to the database.

Configuring migration/env.py to work with the database

In order for Alembic to be able to correctly interact with our database, we need to make changes to the migration/env.py file. This file is responsible for configuring migrations, and in it we need to specify information for connecting to the database, as well as import all the models.

Configuration steps

1. Importing the connection and models

First of all, we import a reference to connect to the database and the Base base class from the database.py file, as well as all our models from the models.py file:

from database import Base, DATABASE_URL
from models import User, Comment, Post, Profile

2. Connection configuration

Next, we tell Alembic which URL to use to connect to the database:

config = context.config
config.set_main_option("sqlalchemy.url", DATABASE_URL)

Here we use the DATABASE_URL variable, which contains the connection string to our PostgreSQL database.

3. Defining Metadata

We tell Alembic where to look for information about models. To do this, we assign the target_metadata variable the metadata from Base, which includes all models:

target_metadata = Base.metadata

These steps prepare the env.py file to work with your database. When we set target_metadata, Alembic gets access to the structure of all our models and uses them to create or update the database schema, adding new tables and columns as the models change. The env.py file thus becomes the glue between your SQLAlchemy models and your migrations in Alembic, allowing you to easily make and track changes to your database.

Changed code migration/env.py

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
from database import Base, DATABASE_URL
from models import User, Comment, Post, Profile

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
# leave the rest of the code unchanged

Preparing the migration file

Now, to prepare the migration file, let's create an instruction for Alembic to use to create the tables. In the terminal, run the following command:

alembic revision --autogenerate -m "Initial revision"

This command will create a migration file that Alembic will use to create our tables based on the models.

If everything was entered correctly, you will see a message like this. Here Alembic reports that it has found the tables and then says that it has generated a file with instructions for migrations at the specified address.

If we look into this file, we will see that each column is described in a more familiar style for SQL CORE. This indicates that we have managed to clearly explain to Alembic what fields we want to get at the output.

Important nuances of working with Alembic and ENUM data types

When using Alembic to manage migrations, there are several important features related to ENUM data types in PostgreSQL that need to be taken into account. Let's look at two main issues and how to solve them.

1. Issues when creating a column with ENUM

Once the ENUM data type is created in PostgreSQL, its name is reserved and considered unique. When working with Alembic, this feature is not always taken into account automatically, which can lead to errors in the future.

Example of a problem

Let's say we have a column like this:

sa.Column('gender', sa.Enum('MALE', 'FEMALE', name='genderenum'), nullable=False)

When you first run the migration, Alembic will correctly create the table and ENUM type for this column. However, if you want to change this ENUM in the future (for example, add a new value), you may encounter an error when running repeated migrations, since Alembic will try to create a genderenum type that already exists in the database.

Solution

To avoid this problem, you need to explicitly set the create_type=False parameter so that Alembic does not try to re-create the ENUM type:

sa.Column('gender', sa.Enum('MALE', 'FEMALE', name='genderenum', create_type=False), nullable=False)

Tip: Always specify create_type=False for ENUM columns to avoid conflicts when re-running migrations.

2. Issues when downgrading migrations

When you drop tables with Alembic, the tables are dropped, but the ENUMs associated with them remain in the database. This can lead to conflicts if you want to use the same names for new ENUM types in the future.

Solution

To make Alembic correctly drop ENUM types when you roll back migrations, you need to extend the downgrade method as follows:

def downgrade() -> None:
    # Deleting tables
    op.drop_table('comments')
    op.drop_table('posts')
    op.drop_table('users')
    op.drop_table('profiles')

    # Removing ENUM types
    op.execute('DROP TYPE IF EXISTS ratingenum')
    op.execute('DROP TYPE IF EXISTS genderenum')
    op.execute('DROP TYPE IF EXISTS professionenum')
    op.execute('DROP TYPE IF EXISTS statuspost')

In this case, when rolling back the migration, not only the tables will be deleted, but also all the corresponding ENUM types.

Important: Use this method only when you need to delete the ENUM types themselves, to avoid unnecessary deletion if the types may be needed in other parts of the application.

Performing Migrations

Now that you have made all the necessary edits and taken into account the peculiarities of working with ENUM, you can start migrating the database.

Updating the database to the latest migration version

To update the database to the latest version, run the command:

alembic upgrade head

This command will apply all migrations that are in the migrations/versions directory and bring the database structure up to date. upgrade head will update the database schema to the latest version that was created and registered in the migration files.

Performing a migration to a specific ID

If you need to update the database not to the latest version, but to a specific migration, you can specify the ID of the desired migration. For example:

alembic upgrade d97a9824423b

In this case, d97a9824423b is a unique ID of the migration you need. Alembic will perform all migrations up to the specified ID, updating the database to the state corresponding to this version.

Performing migration

I see that everything went well. I'll go to the database and check.

All tables are in place.

Rolling Back Migrations: Downgrade

We've already covered the topic of rolling back migrations, and now I want to make sure you've made all the necessary edits so that not only tables are dropped when rolling back, but also their associated ENUM data types.

Sometimes you may need to roll back changes to your database, for example if a new version has introduced bugs or requires additional work. Alembic supports the ability to roll back migrations using the downgrade command, which allows you to return your database to a previous or specific state.

Rollback one version

To rollback a migration one version, use the following command:

alembic downgrade -1

Rollback to a specific migration

If you need to rollback the database to a specific migration, specify the ID of this migration:

alembic downgrade d97a9824423b

This command will revert the database to the state corresponding to the migration with ID d97a9824423b and remove all changes applied after it.

Downgrade() example considering ENUM types removal

Here is what the downgrade method should look like after making all the necessary edits:

def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('comments')
    op.drop_table('posts')
    op.drop_table('users')
    op.drop_table('profiles')
    # ### end Alembic commands ###

    # Removing ENUM types
    op.execute('DROP TYPE IF EXISTS ratingenum')
    op.execute('DROP TYPE IF EXISTS genderenum')
    op.execute('DROP TYPE IF EXISTS professionenum')
    op.execute('DROP TYPE IF EXISTS statuspost')

Breakdown:

Dropping tables: The op.drop_table() commands remove all specified tables from the database. This is automatically generated by Alembic and removes tables based on the current state of migrations.
Dropping ENUM types: We are adding commands to remove ENUM types that may remain in the database even after the associated tables are dropped. The op.execute('DROP TYPE IF EXISTS ...') command removes ENUM types if they exist.

Testing the rollback

Once you have made your edits and are sure you are ready, you can rollback:

Roll back one version:

alembic downgrade -1

Rollback to a specific migration:

alembic downgrade d97a9824423b

This process will delete both tables and associated ENUM types, gracefully reverting the database to its previous states.

We didn't get any errors, so it worked as expected.

Now, with these changes in place, you can confidently manage migration rollbacks, knowing that Alembic will handle both tables and ENUM types in your database gracefully.

Conclusion

Guys, today we have dived deep into the basics of working with SQLAlchemy. We have discussed setting up a database, creating tables, establishing relationships between them, and also looked at the process of migrations using Alembic. I understand that the volume of material turned out to be quite impressive, and it may be difficult to grasp everything at first. But do not worry - practice will help you consolidate your knowledge!

I tried to explain everything as clearly as possible, but the best way to master this topic is to continue working with the code. Take your time, carefully study everything we have talked about, and do not hesitate to consult additional information. This is especially true for working with relationships between tables - without understanding them, it will be difficult to move on.

In the following articles, we will move on to more practical issues: adding, retrieving and filtering data in tables. We will dive deeper into working with relationships between tables so that you can see how this is applied in practice and why setting up relationships is so important.

My plan is to first go through all the intricacies of working with asynchronous SQLAlchemy. We will complete this mini-course by creating a full-fledged project on FastAPI, where we will apply all the knowledge we have gained in practice.

You can easily deploy the finished project, as well as the PostgreSQL database that we posted in this article, on Amverum Cloud in just a few minutes - but we will talk about this separately.

Your response will determine how soon the next article on asynchronous work with SQLAlchemy will be published. Preparing such materials requires time and effort, so any support from you will be very useful!

You can find the full source code used in this article, as well as exclusive content that I do not publish on Habr, in my Telegram channel “Easy Path to Python”. The channel already has almost 1000 participants who study and discuss Python together!

See you soon!

Pydantic 2: The Complete Guide for Python Developers - From Basics to Advanced Techniques

Amverum Cloud — Tue, 03 Jun 2025 05:02:29 +0000

Recently, I noticed that there is very little accessible and understandable information about the Pydantic library on the Russian-language Internet, especially about its updated version 2. This seems strange, because Pydantic is a very easy-to-learn library that will take you only a few hours to learn. However, once you master it, you will have a powerful tool at your disposal that can be effectively used in most Python projects.

Interesting fact: more than 30% of all Python projects use Pydantic, even if it is not always noticeable at first glance. And frameworks such as FastAPI generally build their logic on top of Pydantic, making it an integral part of their solutions.

This is confirmed by the statistics of projects deployed in our cloud Amverum Cloud. The technology is really often used by our users.

About Amverum Cloud

Amverum - is a cloud for easy deployment of applications via git push. Built-in CI/CD, backups and monitoring allow you to deploy a project with three commands in the IDE and not think about setting up the infrastructure. Amverum is simpler than using a VPS or Kubernetes cluster. Git push amverum master, and your bot or site will be launched in the cloud.

As you may have guessed from the title, today we will take a detailed look at how to use Pydantic 2 in your projects, and most importantly — why you need it. We will cover the key concepts, features, and changes that have appeared in the new version of the library.

What you will learn by the end of this article:

What is Pydantic and its main purpose.
The concept of model in Pydantic.
We will study in detail what fields are and how Pydantic's built-in mechanisms help with data validation.
We will examine custom field validation (via field_validator) and global validation at the model level (model_validator).
We will deal with the issue of auto-generated fields in Pydantic
We will dive into model settings using ConfigDict to understand why they are needed and how to use them effectively.
Let's look at the model inheritance mechanism, which can significantly optimize and simplify your code.
Learn how to integrate Pydantic with ORM models (using SQLAlchemy as an example, although this can be applied to other ORMs as well).
Learn how to transform data into convenient formats — dictionaries and JSON strings.

As a result: you will get acquainted with Pydantic version 2 and master its main methods and approaches.

Brief theoretical block

Before moving on to practice, let's get acquainted with the basic concepts and capabilities of Pydantic 2. This block will be theoretical, but extremely important — we will then consolidate all its aspects in practice.

I will try to explain everything in as much detail and "meticulously" as possible, so that there are no questions left.

What is Pydantic?

Pydantic 2 is a Python library for data validation and transformation. It helps developers ensure that input data conforms to established rules and types, and automatically transforms it into the required formats.

Key features of Pydantic:

Data validation: checks input data against expected types and constraints.
Data transformation: automatically transforms data into the required types and formats.

Models in Pydantic

Models in Pydantic inherit from the BaseModel class. Each model describes a set of fields that represent the structure of the data and the conditions for its validation.

Field descriptions:

Typing: Fields in a model are described with types, for example, name: str. This provides basic type validation.
Usage Field(): Allows you to annotate fields with additional parameters such as default values, constraints, and other settings.

Basic Model Example:

from pydantic import BaseModel, Field

class User(BaseModel):
    name: str
    email: str = Field(..., alias='email_address')

Field Validation

Minimal Type Validation: Using Python's built-in types (e.g. str, int), you can perform basic field validation.
Using Validators: Pydantic provides validators such as EmailStr for validating email addresses. Using advanced validators requires installing additional dependencies: pydantic[email] or pydantic[all].

Example with validator:

from pydantic import BaseModel, EmailStr

class User(BaseModel):
    name: str
    email: EmailStr

Decorators in Pydantic

Pydantic 2 adds new capabilities for validating and evaluating fields using decorators.

@field_validator — replaces the old @validator and allows you to add custom field validation logic. Called when a model is created or modified.

Usage example @field_validator:

from pydantic import BaseModel, field_validator

class User(BaseModel):
    age: int

    @field_validator('age')
    def check_age(cls, value):
        if value < 18:
            raise ValueError('Age must be over 18 years')
        return value

@computed_field - a field that is calculated based on other data in the model. It can be used to automatically generate values, as well as for validation.

Example of usage @computed_field:

from pydantic import BaseModel, computed_field

class User(BaseModel):
    name: str
    surname: str

    @computed_field
    def full_name(self) -> str:
        return f"{self.name} {self.surname}"

Working with ORM

Pydantic supports integration with ORM (e.g. SQLAlchemy) for validation and transformation of data retrieved from the database.

To configure the model to work with ORM, use the ConfigDict parameter with the from_attributes=True flag.

Example:

from datetime import date
from pydantic import BaseModel, ConfigDict

class User(BaseModel):
    id: int
    name: str = 'John Doe'
    birthday_date: date

    config = ConfigDict(from_attributes=True)

To create a Pydantic model from an ORM object, use the from_orm method.

Example:

user = User.from_orm(orm_instance)

Methods for working with data

dict() / model_dump() — convert the model into a Python dictionary. In version 2, the model_dump() method became an analogue of dict().

Example:

data = user.model_dump()

json() / model_dump_json() — convert the model to a JSON string. In the new version, the model_dump_json() method replaces the old json().

Example:

json_data = user.model_dump_json()

Passing data to the model

Named arguments: Model fields can be set directly when creating an instance.

Example:

user = User(name="Oleg", age=30)

Unpacked Dictionaries: You can pass field values using dictionary unpacking **.

Example:

user_data = {"name": "Oleg", "age": 30} user = User(**user_data)

By now you should have learned that Pydantic 2 is a powerful data tool that supports validation and transformation of various data types, as well as ORM integration and custom validators.

Now let's put it all into practice.

Getting Started with Pydantic 2

To get started, create a new project in your favorite IDE, such as PyCharm. If you don't have Pydantic installed yet, install it using the following command:

pip install -U pydantic[all]

What does this command do:

-U — updates Pydantic to the latest version if it is already installed, or just installs the latest available version if Pydantic is missing.
[all] — this flag adds all sorts of additional modules and validators that might be useful in the project, such as an email address validator and other advanced features.

Now that the installation is complete, we are ready to start practicing Pydantic 2.

Let's describe the first Pydantic model

from datetime import date
from pydantic import BaseModel


class User(BaseModel):
    id: int
    name: str
    birthday_date: date

In this Pydantic model, we have defined three required fields: id, name and birthday_date. These fields will be validated automatically.

Let's now create an object of the User class, passing the required parameters.

oleg = User(id=1, 
            name='Oleg', 
            birthday_date=date(year=1993, month=2, day=19))

Everything is clear here and we don't see any differences from working with a regular class yet. Now we'll get data in the form of a dictionary and a JSON string, and then I'll show you a few tricks.

To transform a model into a Python dictionary (dict), you can use the dict() method or its full analogue model_dump().

To transform into a JSON string, you can use the json() or model_dump_json() methods.

We'll perform the transformation using each of the methods.

to_dict = oleg.model_dump()
to_json = oleg.model_dump_json()

print(to_dict, type(to_dict))
print(to_json, type(to_json))

Output the object type to the console so that the differences are visible.

What happens if we create another user's object and pass the data as follows?

alex = User(id="2",
            name='Alexey',
            birthday_date="1990-11-22")

The first thing that comes to mind is: "Of course it's a mistake!" Which is logical, because we passed the user ID as a string when an integer is expected, and in the birthday_date field we passed a string instead of a date object, but we'll check.

alex = User(id="2",
            name='Alexey',
            birthday_date="1990-11-22")

to_dict = alex.model_dump()
to_json = alex.model_dump_json()

print(to_dict, type(to_dict))
print(to_json, type(to_json))

With this simple example, I showed how Pydantic automatically converts data into the format we need. This wonderful feature of the library can be incredibly useful in a variety of situations, and I think you can guess which ones.

Now let's create another user and describe it as follows:

dima = User(id="3",
            name=156,
            birthday_date="1990-11-22")

to_dict = dima.model_dump()
to_json = dima.model_dump_json()

print(to_dict, type(to_dict))
print(to_json, type(to_json))

Logically, everything should work correctly now, right? We've already checked that the string is converted to an integer, and the date passed as a string is converted to a date format. Let's make sure of that.

Unexpectedly, we got an error. Let's figure out why.

Why did the error occur?

Pydantic performs data validation and can also automatically convert types in some cases. However, such conversions only occur for data that can be safely and unambiguously cast to the expected type. For example:

If the model expects an int, and we pass the string "123", Pydantic can convert it to an int.
If the model expects a str, but you pass an int, as in our case, Pydantic does not try to perform automatic typecasting, since it is not always obvious that an integer can be correctly interpreted as a string.

There are two ways out of this situation:

Try to pass correct data for ambiguous cases
Use field validators that will perform data transformation inside the model before the main type check and convert it to a string.

Field validator (field_validator) in Pydantic

In previous versions of Pydantic, this decorator was called validate. Now, in new versions, its name has changed to field_validator, which better reflects its purpose.

Purpose of the decorator

The field_validator decorator is used to check the correctness of the fields of the Pydantic model. In addition to validation, it can be used to transform data before saving it to the model.

Importing the decorator

First, we import the necessary elements from Pydantic:

from pydantic import BaseModel, field_validator
from datetime import date

Usage example

Let's create a User model, adding a validator for the name field:

class User(BaseModel):
    id: int
    name: str
    birthday_date: date

    @field_validator('name', mode='before')
    def validate_name(cls, v):
        return str(v)

The field_validator decorator always takes one mandatory argument - the name of the field to validate. The second argument, which is preferable to specify, is mode.

It is important to note the use of the mode='before' parameter. This tells Pydantic to perform validation and data transformation before the model instance is created, not after. Another option is mode='after'.

The method itself takes the field (v) and then starts validating it. The above is a simple example, but we will complicate it very soon.

Now our validator for the name field will automatically convert any values passed to a string.

Important. Most likely, your IDE will complain about the description of cls in the validator method. This is not an error, but in order to avoid an annoying warning, you can describe this type of decorators as follows:

@field_validator('name', mode='before')
@classmethod
def validate_name(cls, v):
        return str(v)

Let's check the model

Let's try to create an instance of the model with the following data:

user_data = {'id': 3, 'name': '156', 'birthday_date': '1990-11-22'}
user = User(**user_data)
print(user.dict())

Result:

{'id': 3, 'name': '156', 'birthday_date': datetime.date(1990, 11, 22)}

There are no errors, the data was successfully converted and serialized to JSON. However, there is one problem here - our validator is used only to convert the data, not to validate it.

Unexpected result

Let's say we pass the following data:

dima = User(
    id="3",
    name=("Kolya", True, False, 0, 19933),
    birthday_date="1990-11-22"
)

Result:

{'id': 3, 'name': "('Kolya', True, False, 0, 19933)", 'birthday_date': datetime.date(1990, 11, 22)}

There is no error, but something is clearly wrong - instead of a string in the name field, we got a tuple that was simply converted to a string. Obviously, this situation is unacceptable.

Validation Fix

To avoid this situation, we can add strict data type checking to the validator:

@field_validator('name', mode='before')
def validate_name(cls, v):
    if isinstance(v, int):
        return str(v)
    elif isinstance(v, str):
        return v
    else:
        raise ValueError("The name must be a string or a number")

Now our validator checks whether the passed value is a string or a number. If the value is not suitable, a ValueError exception is raised.

Checking the fix

Let's try passing the incorrect data again:

dima = User(
    id="3",
    name=("Kolya", True, False, 0, 19933),
    birthday_date="1990-11-22"
)

Result:

An error occurred: 1 validation error for User
name
  Value error, The name must be a string or a number [type=value_error, input_value=('Коля', True, False, 0, 19933), input_type=tuple]
    For further information visit https://errors.pydantic.dev/2.9/v/value_error

Now the validator worked correctly - it stopped the object creation and returned an error indicating that the passed value was incorrect.

Thus, field_validator can be used not only for data transformation, but also for strict validation of its correctness. In this example, we implemented a check that avoids invalid data in the model, and returned an appropriate error if the conditions are not met.

Model validator (model_validator) in Pydantic

In older versions of Pydantic, this decorator was called root_validator. Its main purpose is to validate the model as a whole, after all fields have already been individually validated. This allows complex checks to be performed that depend on several model fields at once.

Key features of the @model_validator decorator:

Executed after individual fields have been validated.
Has access to all model fields at once.
Can change field values or the entire model.
Used for complex validations involving multiple fields.

Example of using model_validator

Let's extend our User class by adding a model validator to check the user's age and set a default name:

from pydantic import BaseModel, field_validator, model_validator
from datetime import date

class User(BaseModel):
    id: int
    name: str
    birthday_date: date

    @field_validator('name', mode='before')
    def validate_name(cls, v):
        if isinstance(v, int):
            return str(v)
        elif isinstance(v, str):
            return v
        else:
            raise ValueError("The name must be a string or a number")

    @model_validator(mode='after')
    def check_age(self):
        today = date.today()
        age = today.year - self.birthday_date.year - (
            (today.month, today.day) < (self.birthday_date.month, self.birthday_date.day))

        if age < 18:
            raise ValueError("User must be over 18 years old")
        if age > 120:
            raise ValueError("Age cannot exceed 120 years")
        return self

    @model_validator(mode='after')
    def set_default_name(self):
        if self.name.strip() == '':
            self.name = f"User_{self.id}"
        return self

In this example:

The check_age method checks that the user's age is greater than 18 but less than 120. This check requires access to the birthday_date field and the current date, so it is implemented as a model validator.
The set_default_name method sets the default name if the name field is empty. This validator uses multiple fields (name and id), so it is also implemented at the model level.

Both validators use the after mode, which means that they are executed after the individual fields have been validated.

Example usage:

try:
    user = User(id=1, name="John", birthday_date=date(2000, 1, 1))
    print(user)
except ValueError as e:
    print(f"Error: {e}")

try:
    user = User(id=2, name="", birthday_date=date(2010, 1, 1))
    print(user)
except ValueError as e:
    print(f"Error: {e}")

try:
    user = User(id=3, name="Alice", birthday_date=date(1900, 1, 1))
    print(user)
except ValueError as e:
    print(f"Error: {e}")

This example demonstrates how @model_validator helps to perform complex checks and modify the model after validating individual fields.

Computed fields (computed_field)

The @computed_field decorator allows you to create fields that are calculated "on the fly" when accessed. This is useful when you need to automatically get values based on other model fields.

An example with computed fields:

Let's add the full_name and age computed fields to our User class:

from pydantic import BaseModel, computed_field
from datetime import date
from dateutil.relativedelta import relativedelta

class User(BaseModel):
    id: int
    name: str
    surname: str
    birthday_date: date

    @computed_field
    def full_name(self) -> str:
        return f"{self.name} {self.surname}"

    @computed_field
    def age(self) -> str:
        today = date.today()
        delta = relativedelta(today, self.birthday_date)
        return f"{delta.years} years, {delta.months} months и {delta.days} days"

The full_name field is calculated by concatenating the first and last names.
The age field calculates the user's age in years, months, and days using the relativedelta module library of the dateutil library.

An example of using calculated fields:

alex = User(id=1, name="Alexey", surname="Yakovenko", birthday_date="1993-02-19")
print(alex.dict())

Result:

{'id': 1, 'name': 'Alexey', 'surname': 'Yakovenko', 'birthday_date': datetime.date(1993, 2, 19), 'full_name': 'Alexey Yakovenko', 'age': '31 years, 7 months and 28 days'}

As you can see, computed fields work automatically, providing convenient access to derived values.

Decorators @model_validator, @field_validator and @computed_field allow you to flexibly and efficiently manage data validation in Pydantic models, as well as add computed fields. Model-level validation is useful for complex checks, while computed fields make it easier to work with derived values without requiring additional logic in the code.

Computed fields can be combined with model_validate. In this format. We create a new field and then, using model_validate, we can check whether everything is correct. In addition, the check can be done at the stage of creating the computed field itself.

To consolidate this block, practice using decorators yourself: model_validate, field_validate and computed_field.

Annotable fields in Pydantic 2: using the Field function

Pydantic 2 offers developers a convenient and powerful way to work with data model fields, and one of the key tools for this is the Field function. It allows you to detail the behavior of fields: set default values, add metadata, configure validation, and even document the model. In this block, I will talk about how to use Field to configure models in Pydantic 2.

Import the necessary modules:

from pydantic import BaseModel, Field

What is a Field in Pydantic 2?

The Field function allows you to add metadata and settings to model fields that Pydantic uses for validation, serialization, and documentation. Here are the main parameters that can be passed to Field:

default: sets the default value for the field.
default_factory: a function that returns the default value.
alias: an alternative name for the field for serialization and deserialization.
title: the title of the field for documentation.
description: the description of the field for documentation.
exclude: excludes the field from serialization.
repr: determines whether the field will be included in the string representation of the model.

A simple example of using Field

Let's say we have a User model where we want to set default values and add some metadata:

class User(BaseModel):
    id: int = Field(default=1, description="Unique user identifier")
    name: str = Field(default="John Doe", title="Username", description="Full name")
    role: str = Field(default="user", alias="user_role", description="User role in the system")

In this example:

The id field has a default value of 1 and a description for documentation.
The name field describes the user name with a title for documentation.
The role field has an alias user_role that will be used when serializing and deserializing data.

Field Validation

The Field function also allows you to set various restrictions on values. For example, you can define minimum and maximum values for numeric fields, or limit the length of strings.

Validation example via Field

class Product(BaseModel):
    price: float = Field(gt=0, description="The price must be greater than zero")
    name: str = Field(min_length=2, max_length=50, description="Product name must be between 2 and 50 characters")

Here:

The price field must be greater than zero.
The name field has string length restrictions: minimum 2 characters, maximum 50.

Here is a detailed description of commonly used built-in validators:

1. gt, ge, lt, le — for numeric restrictions

These validators are used to restrict numeric values (integers, floats, and other types that support arithmetic).

gt (greater than): Checks that the value is greater than the specified number.

Example: Field(gt=0) — the value must be greater than zero.

ge (greater than or equal): Checks that the value is greater than or equal to the specified number.

Example: Field(ge=1) — the value must be at least one.

lt (less than): Checks that the value is less than the specified number.
Example: Field(lt=100) - the value must be less than one hundred.
le (less than or equal): Checks that the value is less than or equal to the specified number.

Example: Field(le=10) - the value must be no greater than ten.

These validators are useful for specifying ranges of values, such as checking age, price, rating, and any other numeric data that has upper and lower bounds.

Example with numeric constraints:

class Product(BaseModel):
    price: float = Field(gt=0, le=10000, description="The price must be positive and not exceed 10,000")
    rating: int = Field(ge=1, le=5, description="Rating must be from 1 to 5")

Where:

The price field must be greater than 0 and less than 10,000.
The rating field must be between 1 and 5.

2. max_length, min_length — for string fields

These validators are used to limit the length of strings, which is important for validating text data such as usernames, descriptions, and other fields.

min_length: Specifies the minimum number of characters that a string must contain.

Example: Field(min_length=3) — the string must be at least 3 characters long.

max_length: Specifies the maximum number of characters a string can contain.

Example: Field(max_length=100) — the string must contain no more than 100 characters.

An example of using string length validators:

class User(BaseModel):
    username: str = Field(min_length=3, max_length=20, description="Username must be between 3 and 20 characters long")
    bio: str = Field(max_length=300, description="Profile description must not exceed 300 characters")

Here:

The username field must be between 3 and 20 characters long.
The bio field is limited to 300 characters.

3. regex — for regular expression validation

This validator allows you to check string values for compliance with a regular expression. Regular expressions (regex) allow you to flexibly describe acceptable string formats — for example, to validate email addresses, phone numbers, date formats, etc.

regex: Specifies a regular expression that the string must match. Example: Field(regex=r"[^@]+@[^@]+.[^@]+") — checks that the string is a valid email address.

Example of using regex:

class User(BaseModel):
    email: str = Field(regex=r"[^@]+@[^@]+.[^@]+", description="The email must be in the correct format")
    phone_number: str = Field(regex=r"^\+\d{1,3}\s?\d{4,14}$", description="The phone number must be in the format +123456789")

Here:

The email field must match the email address pattern (characters before @, domain name, dot and domain).
The phone_number field must match the international phone number format.

Dynamic default values

Sometimes you need to generate values for fields dynamically. This is done with the default_factory parameter, which takes a function to generate the value.

Example with default_factory

from uuid import uuid4


class Item(BaseModel):
    id: str = Field(default_factory=lambda: uuid4().hex)

Here, each id field will automatically receive a unique identifier when a new Item model object is created.

Using aliases

For compatibility with external APIs or for code readability, you can define aliases for fields. An alias is an alternative name for a field that will be used during serialization or deserialization.

Example with aliases

class User(BaseModel):
    username: str = Field(alias="user_name")

When you specify an alias for a field, the field will have one name in your Python code, but when you send or read data as JSON, it will use the name specified as the alias.

In this example, the username field will be serialized and deserialized as user_name.

Excluding fields from serialization

Sometimes you want to hide certain fields when serializing data — for example, to avoid sending sensitive information like passwords.

Example of excluding fields

class User(BaseModel):
    password: str = Field(exclude=True)

In this example, the password field will be excluded from the serialized representation of the model object.

This means that when serializing the model object (for example, when converting it to JSON or a dictionary), the password field will not be included in the result. This way, you can hide sensitive information such as passwords from external systems or clients.

Customizing the string representation of the model

You can control which fields will be displayed in the string representation of the model using the repr parameter.

Example of customizing the representation

class Config(BaseModel):
    debug_mode: bool = Field(repr=False)

The string representation of a model object is what is returned when calling repr() or str(). For example, when you want to see the state of an object or debug code, you can print its string representation.

The repr=False parameter allows you to exclude certain fields from this representation, so that they are not displayed when printing the object, even if they are present in the model.

In our example, the debug_mode field will not be displayed in the string representation of the model, which can be useful for hiding technical information.

Advanced capabilities via Annotated

Now that we have covered the basic capabilities of Field, we can move on to using annotations via Annotated. This method allows you to add metadata and validation more flexibly and precisely.

Annotated example

from typing_extensions import Annotated

class User(BaseModel):
    id: Annotated[int, Field(gt=0)]
    name: Annotated[str, Field(min_length=2, max_length=50)]
    email: Annotated[str, Field(regex=r"[^@]+@[^@]+.[^@]+")]
    role: Annotated[str, Field(default="user")]

Where:

The id field must be greater than zero.
The name field is limited to 2-50 characters.
The email field must match a regular expression that validates the email format.
The role field has a default value of "user".

The current preferred approach is to use Annotated in the field details.

The Field feature in Pydantic 2 provides developers with flexible tools for customizing model fields. It allows fine-tuning validation, setting default values, using aliases, and adding metadata for documentation. Using Annotated makes this process even more powerful and convenient.

These features help create structured and secure data models, providing easy integration with external APIs and clear validation of incoming data.

Model Configuration in Pydantic 2

In Pydantic 2, model configuration is now specified via ConfigDict rather than the old Config class format. This is a major change that simplifies and makes configuration more flexible.

What it looks like now:

Instead of writing:

class MyModel(BaseModel):
    class Config:
        from_attributes = True

now we use ConfigDict:

from pydantic import BaseModel, ConfigDict

class MyModel(BaseModel):
    model_config = ConfigDict(from_attributes=True)

Main ConfigDict options

from_attributes=True - allows you to create a model object directly from Python object attributes (e.g. when model fields match attributes of another object). Most often, this option is used to convert ORM models to Pydantic models.
str_to_lower, str_to_upper - convert all model strings to lower or upper case
str_strip_whitespace - whether to remove leading and trailing whitespace for str types (similar to strip)
str_min_length, str_max_length - sets the maximum and minimum string length for all string fields
use_enum_values - whether to populate models with values chosen from enums instead of using raw values? This is often needed when working with ORM models where columns are defined as enums (ENUM).

Why is this important?

Because it makes model configuration more explicit, convenient, and flexible. Using ConfigDict allows you to define model parameters directly through a dictionary, avoiding the need to create nested classes, which simplifies the configuration process and makes it more intuitive. This is especially useful when working with large and complex data models that require a high degree of customization.

Now you can easily define and change model parameters without diving into cumbersome hierarchical constructs. This contributes to better readability and maintainability of the code, speeding up the development process.

For a more detailed study of all ConfigDict parameters and methods, I recommend referring to the official documentation where you will find a full description of the capabilities and examples of use.

Inheritance in Pydantic 2

Inheritance in Pydantic allows you to create models that can override or extend attributes and methods of their parent models. This makes your code more flexible and helps avoid duplication.

An example of how inheritance works in Pydantic:

from pydantic import BaseModel


class ParentModel(BaseModel):
    name: str
    age: int


class ChildModel(ParentModel):
    school: str


parent = ParentModel(name="Alex", age=40)
child = ChildModel(name="Bob", age=12, school="Greenwood High")

In this example, ChildModel inherits the name and age fields from ParentModel and adds its own school field.

Advantages:

Code reuse: Common fields and methods can be defined in the base model.
Extensibility: Child models can add new fields or methods.

In Pydantic, when you create a base class with settings, those settings and attributes can be inherited by child classes. This avoids code duplication and makes it easier to maintain consistency.

For example, you can define common parameters in the base model and then use inheritance to create specialized models that automatically receive all the properties and settings of the base class, but can add their own unique attributes. This approach makes the code cleaner and more manageable.

Pydantic 2 with ORM using SQLAlchemy as an example

If you follow my publications on Habr, then you know about the series of articles dedicated to working with the SQLAlchemy Python framework. In the previous article "Asynchronous SQLAlchemy 2: Step-by-Step Guide to Session Management, Adding and Retrieving Data with Pydantic" we looked at the integration of SQLAlchemy with Pydantic in detail

Today I will briefly talk about the key concepts that explain why SQLAlchemy needs Pydantic and give a practical example.

Why does SQLAlchemy need Pydantic?

When working with SQLAlchemy in ORM style, we encounter one significant inconvenience: data is returned as table model objects, which is not always convenient for further processing. Developers usually prefer to work with data in JSON or Python dictionaries. This is where Pydantic comes to the rescue.

The principle of SQLAlchemy and Pydantic integration

The process of SQLAlchemy and Pydantic integration can be described in the following steps:

Describing the table model in SQLAlchemy
Creating a Pydantic model to work with the received data
Querying data from the table via SQLAlchemy
Converting a SQLAlchemy object to a Pydantic object
Using the model_dump or model_dump_json methods to obtain data in the required format

Practical example

Let's look at an example of SQLAlchemy and Pydantic integration:

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from pydantic import BaseModel, ConfigDict


# Step 1: Describing the SQLAlchemy Model
Base = declarative_base()


class UserORM(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)


# Step 2: Create a Pydantic Model
class UserPydantic(BaseModel):
    id: int
    name: str
    email: str

    model_config = ConfigDict(from_attributes=True)


# Setting up a database and session
engine = create_engine("sqlite:///example.db")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)


# Step 3: Request data
def get_user(user_id: int):
    with Session() as session:
        user = session.select(UserORM).filter_by(id = user_id).scalar_one_or_none()

        # Step 4: Convert SQLAlchemy Object to Pydantic
        if user:
            user_pydantic = UserPydantic.from_orm(user)

            # Step 5: Getting the data in the right format
            return user_pydantic.dict()
        return None


# Example of use
user_data = get_user(1)
print(user_data)

Here we see the use of from_orm() methods to create a Pydantic model from an ORM object and dict() to convert the Pydantic model to a dictionary.

Modern approach

With new versions of Pydantic (especially 2.0+), the following style is recommended:

def get_user(user_id: int):
    with Session() as session:
        user = session.select(UserORM).filter_by(id = user_id).scalar_one_or_none()

        # Step 4: Convert SQLAlchemy Object to Pydantic
        if user:
            user_pydantic = UserPydantic.model_validate(user)

            # Step 5: Getting the data in the right format
            return user_pydantic.model_dump()
        return None

Key changes and rationale

from_orm() → model_validate()

from_orm() is now an alias for model_validate().
model_validate() is more generic and can be used with more than just ORM objects.
It is recommended to use model_validate() for greater flexibility and compliance with modern Pydantic standards.

dict() → model_dump()

dict() has become an alias for model_dump().
model_dump() provides more customizable output.
Using model_dump() makes code more explicit and compliant with the new Pydantic API.

While both code options are functional, switching to the new methods (model_validate() and model_dump()) is recommended for the following reasons:

Compliance with modern Pydantic standards and recommendations.
Improved code readability and explicit intent.
Ability to use additional options available in the new methods.
Preparing code for future library updates.

Using from_attributes in model_validate

In Pydantic 2.0+, the model_validate method has become more flexible and convenient. You can directly specify from_attributes=True when calling the method:

user_pydantic = UserPydantic.model_validate(user, from_attributes=True)

This allows you to dynamically control whether object attributes will be used to create a model, without having to change the configuration of the model itself.

Other useful attributes of model_validate

In addition to from_attributes, the model_validate method has several other useful attributes:

strict: bool | None

When set to True, enforces strict type validation.
Example: model_validate(data, strict=True)

context: Any | None

Allows you to pass additional context to validators.
Example: model_validate(data, context={'user_id': 123})

from_attributes: bool | None

As we have already discussed, allows you to extract data from object attributes.

Usage example

class User:
    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age


class UserModel(BaseModel):
    name: str
    age: int


user = User("Alice", 30)


# Using from_attributes
user_model = UserModel.model_validate(user, from_attributes=True)


# Using strict and context
data = {"name": "Bob", "age": "25"}
user_model = UserModel.model_validate(
    data,
    strict=True,
    context={"source": "external_api"},
    from_attributes=False
)

These attributes provide flexibility in data validation, allowing you to tailor the process to the specific requirements of your application.

Using from_attributes directly in model_validate is especially convenient when you need to dynamically switch between different data sources without changing the model configuration.

The concept of reverse validation

The idea is to use Pydantic models not only for output validation, but also for structuring the input parameters of database queries. This ensures type safety and ease of use when forming filters for queries.

Example implementation

1. Defining a Pydantic model for filters

from pydantic import BaseModel, ConfigDict


class TelegramIDModel(BaseModel):
    telegram_id: int

    model_config = ConfigDict(from_attributes=True)

2. Method for searching in the database

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError


class BaseRepository:
    @classmethod
    async def find_one_or_none(cls, session: AsyncSession, filters: BaseModel):
        filter_dict = filters.model_dump(exclude_unset=True)
        try:
            query = select(cls.model).filter_by(**filter_dict)
            result = await session.execute(query)
            return result.scalar_one_or_none()
        except SQLAlchemyError as e:
            # Error handling
            raise

3. Usage

async def get_user_by_telegram_id(session: AsyncSession, telegram_id: int):
    filters = TelegramIDModel(telegram_id=telegram_id)
    user = await UserRepository.find_one_or_none(session, filters)
    return user

By the way, in my universal template for creating Telegram bots based on Aiogram 3 and SQLAlchemy, I used a more complex example. If you are interested in the source code of this template and other exclusive content that I do not publish on Habr, you can get it for free in my Telegram channel "Easy way to Python".

Advantages of this approach

Type safety: Pydantic provides input data type validation.
Flexibility: It is easy to create different filter models for different queries.
Readability: The code becomes more understandable and structured.
Reusability: Filter models can be used in different parts of the application.

This approach demonstrates how Pydantic can be effectively used not only for output data validation, but also for structuring query input parameters. This creates an additional level of abstraction between the business logic and the data access layer, which improves the overall architecture of the application.

This method is especially useful in large projects where consistency and type safety when working with a database must be ensured.

Practical application of Pydantic in various areas

Pydantic is a powerful tool that finds wide application in various areas of development. Here are some key areas where Pydantic can be especially useful:

1. Web development

Pydantic is often used in web frameworks such as FastAPI and Flask to validate incoming data from users. This allows developers to easily process JSON requests and ensure that the data matches the expected types.

On Habr, I have published more than ten large articles about developing your own API using FastApi. If you are interested in this topic, I recommend that you take a look at my profile. There you will find many examples of how to use FastApi together with Pydantic.

2. API and microservices

When creating RESTful APIs, Pydantic helps define data schemas and automatically generate documentation. This simplifies integration between different services and ensures data consistency.

3. Configuration processing

Pydantic can be used to work with configuration files (for example, JSON or YAML). This allows you to easily load and validate application settings, minimizing the likelihood of errors.

4. Data processing

In projects related to data analysis or machine learning, Pydantic helps validate input data and ensure that it meets the necessary requirements before processing it.

5. Testing

Pydantic can be useful when writing tests, allowing you to create dummy data with a guarantee of its correctness. This simplifies the testing process and increases its reliability.

Conclusion

Friends, today we have taken an exciting journey into the world of Pydantic 2, and I hope that this article has become a useful and informative guide for you. Let's briefly summarize:

We have covered the basics of Pydantic and its importance in the Python ecosystem.
We have studied the key concepts of Pydantic 2: models, fields, validation, ConfigDict and inheritance.
We have looked at the integration of Pydantic with ORM, in particular with SQLAlchemy.
We have discussed the practical application of Pydantic in various areas of development.
Noted the performance and functionality improvements of Pydantic 2.

Remember that to fully master Pydantic 2, you need to practice. Experiment with the code and integrate Pydantic into your projects.

If you liked the article, don't forget to like and leave a comment. Your opinion is very important to me.

For even more exclusive content, join my Telegram channel "Easy Path to Python". And if you need to deploy a project on a server and deliver updates to it with three commands in the IDE, register in Amverum Cloud, and get $1 for testing the functionality.

Thank you for your time! Good luck with your projects, and may Pydantic 2 become your reliable assistant in the world of Python development!

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

Amverum Cloud — Sat, 31 May 2025 09:19:32 +0000

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:

Preparing a file with settings (the code will be provided in the article).
Creating a project in Amverum.
Uploading bot files and settings (manually on the website or via GIT).
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:

Open BotFather in Telegram.
Enter the command /newbot.
Specify the bot name (it can be in Russian).
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.

Log in to your bot via BotFather.
Select the Payments section.

Click on Yukassa and select the Connect Test Yukassa option.
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:

Return to BotFather.
Open your bot settings and select Payments.
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

In your favorite IDE, create a new project for the bot.
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&gt;=2.4.1,&lt;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

Imports

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

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.

Initializing settings

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

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.

Logging

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

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:

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:

Adding the project root directory to the system path for correct import of modules.
Specifying the URL for connecting to the database.
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?

The user initiates an action through the bot.
The middleware creates and opens a session.
The necessary operations with the database are performed in the processing function.
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:

Open a session before handling an event.
Pass the session to data — a special dictionary used in Aiogram to pass data between middlewares and handlers.
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:

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

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:

Opens a session via async_session_maker().
Adds the session to the data for subsequent use in handlers.
When processing is complete:

If DatabaseMiddlewareWithCommit is used, a commit is performed.
If DatabaseMiddlewareWithoutCommit is used, changes to the database remain unprocessed.

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:

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.

title:

Invoice title.
Briefly describes the product or service.

description:

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

payload:

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

provider_token:

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

currency:

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

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

reply_markup:

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

How it works:

The user clicks the "Buy" button, calling the handler.
The bot sends the invoice using send_invoice.
Telegram displays the invoice with the "Pay" button.
The user finishes entering payment details.
Telegram sends your bot a pre_checkout_query event to confirm the order. The bot must handle this event within 10 seconds.
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

The send_invoice method sends an invoice to the user.
The pre_checkout_query handler confirms that the payment is ready to be completed.
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

Registration: If you do not have an account yet, register in Amverum Cloud. New users receive a welcome bonus of 1$ to their account.
Creating a project: In the control panel, go to the projects section and click "Create a project".
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.
Uploading files: Select the method for uploading application files. Beginners are recommended to use the Amverum interface, more experienced users - GIT commands.
Check settings: On the next screen, make sure all settings have loaded correctly and click "Finish".
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!

Websockets on FastAPI: Implementing a simple chat with rooms in 20 minutes

Amverum Cloud — Thu, 29 May 2025 18:07:55 +0000

In the near future, I plan to publish a large article on Habr, where I will analyze in detail the development of an anonymous chat in the Telegram MiniApp format. We will create a service for face-to-face communication, which will work inside Telegram and provide instant messaging.

Project technology stack:

FastAPI + WebSocket – for real time and messaging
Redis – for quick search and connection of interlocutors
PostgreSQL – for storing messages and user information
Vue 3 + Pinia – for a convenient and responsive interface
Telegram Mini Apps API – to embed the chat directly into Telegram

In addition to the technical analysis, in the article I will also touch on the issue of monetization. One of the most effective ways to monetize Telegram MiniApp is advertising. This allows the service to remain free for users, and the developer to earn money on the popularity of the project.

As an advertising platform, I will consider RichAds - they offer excellent tools for integrating advertising into Telegram MiniApps, which makes them a logical choice for such a project.

But before we dive into MiniApp development, let's get the basics down. In this article, we will get acquainted with web sockets, learn how they work, and learn how to implement them on FastAPI.

This is only the first part of a large project, which I will refer to in future material to make it more structured and easier to read.

So, what is WebSocket and how can it help us in developing a chat? Let's figure it out.

WebSocket: principles of operation

Imagine a phone call: you dial a number (establish a connection) once, and after that you can freely talk without having to dial the number again for each phrase. WebSocket works in a similar way - it is a technology that creates a permanent "communication channel" between the user's browser and the server, allowing them to exchange messages in real time.

Let's get some terminology straight:

The client is what the person uses: a web page in a browser, an application on a phone or computer
The server is a remote computer that stores and processes our application data

Key advantages of WebSocket:

Persistent connection: like a phone call – establish a connection once and communicate
Fast operation: no time is wasted on reconnection
Two-way communication: both the client and the server can start communicating first
Resource saving: service information is transmitted only at the beginning of the connection

Comparison of approaches to implementing a chat

Classic method (HTTP):

Imagine that you send letters:

Write a message and send it to the server
The server saves the letter
Other chat participants must constantly check the "mailbox" (make requests to the server)
Only after checking will they see the new message

Even if the check is done automatically (for example, via AJAX), you still have to constantly "look into the mailbox", which creates an extra load.

Disadvantages:

The server gets tired of constant checks
Messages arrive with a delay
A lot of Internet traffic is spent

WebSocket Method:

Now imagine a group call:

All participants are connected to the general conversation
When someone speaks, everyone hears it at once
No need to constantly check for new messages
Communication happens instantly

Advantages:

Messages arrive instantly
The server is less loaded
Internet traffic is saved
True real-time communication

This approach works great not only in chats, but also in online games, collaborative document editors and anywhere where an instant response to user actions is needed.

Next, we will create a simple chat on FastAPI to see how it works in practice. For simplicity of the example, we will make both the server and client parts in one application, although in a real "Tet-a-Tet" chat the interface will be a separate application on VueJS3.

What project are we going to develop?

Today we will create a full-fledged FullStack application – a group chat, in which users will be able to:

Create rooms for communication
Join existing rooms
Exchange messages in real time

All participants in the same room will instantly receive new messages without having to refresh the page.

Tech stack

To implement the project, we will use modern and convenient tools:

Python + FastAPI + WebSockets – for the server part and organizing real time
JavaScript – for establishing connections via web sockets and page dynamics
TailwindCSS – for fast and convenient interface styling
HTML + Jinja2 – for rendering pages with data
Amverum Cloud – for fast and easy deployment

Development stages

1. Server part (Backend)

Developing a class for managing web sockets
Creating an endpoint for handling connections via WebSocket
Implementing routes (endpoints) for rendering HTML pages

2. Client part (Frontend)

Developing two HTML pages
Adding JavaScript logic for connecting to WebSocket
Implementing dynamic interface updates

3. Project deployment

A group chat is pointless without external access, so in conclusion we will deploy the project on the Internet.

For deployment, we will use Amverum Cloud – a service that allows you to deploy a FastAPI application in just a couple of minutes. The advantages of this solution:

Automatic HTTPS certificates
Free domain name
Webhooks support (for example, for integration with Telegram bots)

Thus, as a result of the work, we will have a ready, expanded and working group chat that can be used, tested and expanded.

Project preparation

Let's start with project preparation. Let me remind you that we will be writing in the Python FastAPI framework, which is great for implementing WebSocket due to its asynchronous nature and simple API.

Open IDE and create a new project
Create a requirements.txt file and fill it in as follows:

fastapi==0.115.8 # The web framework itself
websockets==15.0 # Library for working with WebSocket
uvicorn==0.34.0. # ASGI server for running the application
jinja2==3.1.5    # Template engine for rendering HTML

Install libraries:

pip install -r requirements.txt

Prepare the project structure:

my_chat_project/
├── requirements.txt      # Project dependencies file
├── app/                  # Main application directory
│ ├── templates/          # Directory with HTML templates
│ │ ├── home.html         # Home page template
│ │ └── index.html        # Main application template
│ ├── static/             # Directory with static files (JS, CSS)
│ │ └── index.js          # JavaScript for index.html
│ ├── api/                # Directory with API routes
│ │ ├── router_page.py    # Routes for pages (HTML)
│ │ └── router_socket.py  # Routes for WebSocket connections
│ └── main.py             # Main application file (launch)

Writing the server part

We will describe the main logic of the server part of our application in the router_socket.py file. This file will be responsible for managing WebSocket connections and sending messages between users.

Importing the necessary modules

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from typing import Dict

FastAPI — used to create a web application.
WebSocket, WebSocketDisconnect — classes for working with WebSocket connections. The first allows you to establish connections, the second — to handle their termination.
Dict — a data type for convenient annotation of the structure for storing user connections.

Initializing the router

FastAPI uses routers (APIRouter) to organize endpoints into logical blocks. In this case, let's create a router with the prefix /ws/chat:

router = APIRouter(prefix="/ws/chat")

This route will handle all WebSocket requests related to the chat.

Creating a connection manager

For convenient connection management, let's create a ConnectionManager class. It will be responsible for connecting and disconnecting users, as well as for sending messages.

Full class code:

class ConnectionManager:
    def __init__(self):
        # Store active connections as {room_id: {user_id: WebSocket}}
self.active_connections: Dict[int, Dict[int, WebSocket]] = {}

    async def connect(self, websocket: WebSocket, room_id: int, user_id: int):
        """
        Establishes a connection with the user.
        websocket.accept() — confirms the connection.
        """
        await websocket.accept()
        if room_id not in self.active_connections:
            self.active_connections[room_id] = {}
        self.active_connections[room_id][user_id] = websocket

    def disconnect(self, room_id: int, user_id: int):
        """
        Closes the connection and removes it from the list of active ones connections.
        If there are no more users in the room, deletes the room.
        """
        if room_id in self.active_connections and user_id in self.active_connections[room_id]:
            del self.active_connections[room_id][user_id]
            if not self.active_connections[room_id]:
                del self.active_connections[room_id]

    async def broadcast(self, message: str, room_id: int, sender_id: int):
        """
        Broadcasts a message to all users in the room.
        """
        if room_id in self.active_connections:
            for user_id, connection in self.active_connections[room_id].items():
                 message_with_class = {
                    "text": message,
                    "is_self": user_id == sender_id
                }
                await connection.send_json(message_with_class)

Code breakdown

Class constructor

self.active_connections is a dictionary that stores active connections grouped by rooms (room_id).
Each room (room_id) stores connected users as {user_id: WebSocket}.

connect

Accepts a WebSocket connection, a room id (room_id), and a user (user_id).
Acknowledges the connection (websocket.accept()).
Adds the WebSocket to self.active_connections.

disconnect

Removes the user's WebSocket from self.active_connections.
If there are no users left in the room, deletes the room.

broadcast

Sends a message to all users in the room.
Additionally adds the is_self flag so that the client can visually highlight its messages.

Initializing the connection manager

Let's create an instance of the ConnectionManager class, which we will use later:

manager = ConnectionManager()

Creating a WebSocket endpoint

Now let's create a WebSocket endpoint that will manage user connections and message transmission in the chat.

@router.websocket("/{room_id}/{user_id}")
async def websocket_endpoint(websocket: WebSocket, room_id: int, user_id: int, username: str): 
    await manager.connect(websocket, room_id, user_id) 
    await manager.broadcast(f"{username} (ID: {user_id}) has joined the chat.", room_id, user_id) 

    try: 
        while True: 
            data = await websocket.receive_text() 
            await manager.broadcast(f"{username} (ID: {user_id}): {data}", room_id, user_id) 
    except WebSocketDisconnect: 
        manager.disconnect(room_id, user_id) 
        await manager.broadcast(f"{username} (ID: {user_id}) has left the chat.", room_id, user_id)

Code breakdown

Route

Endpoint accepts three parameters from URL: room_id, user_id, username.
Each user connects via URL /ws/chat/{room_id}/{user_id}.

Connection

await manager.connect(...) — adds user to active connections list.
await manager.broadcast(...) — notifies all users of the room about new member.

Receiving and sending messages

Infinite loop (while True) listens for incoming messages via websocket.receive_text().
After receiving message, it is broadcast to all users of the room via manager.broadcast(...).

Disconnection

If connection is interrupted (WebSocketDisconnect), manager.disconnect(...) is called.
Sends a message to the chat about the user leaving.

In this section, we created the server part of the chat on WebSocket using FastAPI. We:

Figured out imports and routers.
Created ConnectionManager to manage connections.
Wrote a WebSocket endpoint to receive and send messages.

In the next section, we will look at how to connect the client part and test WebSocket connections.

Writing the client part

The client part will conditionally consist of two main stages:

Description of endpoints for rendering HTML pages
Writing the frontend part itself: HTML + CSS + JS

In this section, we will describe the endpoints that will serve HTML pages. We will implement them in the app/router_page.py file.

Imports

First, import the necessary modules:

from fastapi import APIRouter, Request, Form
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
import random

Let's figure out why we need each import:

APIRouter – allows you to organize routing in the application.
Request – a request object that is passed to the template during rendering.
Form – used to process data transmitted via an HTML form.
Jinja2Templates – needed to work with HTML templates.
HTMLResponse – specifies that the endpoint returns an HTML page.
Random - to generate a random user ID

Initializing the route and renderer

Before describing the endpoints, let's create a templates object that will indicate where the HTML templates are stored, and initialize the router:

templates = Jinja2Templates(directory='app/templates')
router = APIRouter()

Endpoints description

Now let's create two endpoints. The first one will be responsible for rendering the main page, and the second one - for the room page (our group chat).

Endpoint for the main page

@router.get("/", response_class=HTMLResponse)
async def home_page(request: Request):
    return templates.TemplateResponse("home.html", {"request": request})

This endpoint returns the main page home.html. It will contain a form for entering the chat, which we will analyze later.

Endpoint for joining chat

@router.post("/join_chat", response_class=HTMLResponse)
async def join_chat(request: Request, username: str = Form(...), room_id: int = Form(...)):
    # Simple user_id generation
    user_id = random.randint(100, 100000)
    return templates.TemplateResponse("index.html",
                                      {"request": request,
                                       "room_id": room_id,
                                       "username": username,
                                       "user_id": user_id}
                                      )

This endpoint does several things:

Gets username and room_id from the form (using Form(...)).
Generates a random user ID in the range from 100 to 100000
Returns the HTML page index.html with the parameters passed to the template:

room_id – ID of the room the user enters.
username – username.
user_id – generated user ID.

Why is Form(...) used?

FastAPI has several ways to pass data to an endpoint. In this case, Form(...) specifies that the username and room_id parameters are passed through an HTML form using the POST method. This is convenient for processing data entered by the user on a web page.

If we were passing data in the URL (as query parameters), we would have to use Query(...), and if we were passing JSON – Body(...).

At this point, we have endpoints ready for working with HTML pages. In the next section, we will implement the frontend: create HTML templates, styles, and connect WebSocket.

Frontend implementation: HTML, CSS, and JavaScript

Now that we have the server part and endpoints for rendering pages, let's move on to creating the user interface. Our client part will include:

HTML - for the page structure,
CSS (TailwindCSS) - for styling,
JavaScript - for event handling and working with WebSocket.

1. HTML structure

We will create two main HTML files:

home.html – the main page with the chat login form.
index.html – the chat page where real-time communication will take place.

Home page (home.html)

<!DOCTYPE html>
<html lang="en">
<head> 
    <meta charset="UTF-8"> 
    <meta name="viewport" content="width=device-width, initial-scale=1.0"> 
    <title>Login to chat</title> 
    <script src="https://cdn.tailwindcss.com"></script>
</head>
<body class="bg-gray-100 flex flex-col items-center justify-center min-h-screen p-4">
<h1 class="text-3xl font-bold mb-6">Welcome to the chat</h1>

<form action="/join_chat" method="post" class="bg-white p-6 rounded-lg shadow-md w-full max-w-md"> 
    <label class="block" text-gray-700">Enter your name:</label> 
    <input type="text" name="username" required 
            class="w-full p-2 border border-gray-300 rounded-lg mt-2 focus:ring-2 focus:ring-blue-500"> 

    <label class="block text-gray-700 mt-4">Enter room ID:</label> 
    <input type="number" name="room_id" required min="1" 
           class="w-full p-2 border border-gray-300 rounded-lg mt-2 focus:ring-2 focus:ring-blue-500"> 
    <button type="submit" class="w-full bg-blue-500 text-white px-4 ru-2 rounded-lg mt-4 hover:bg-blue-600">Log in 
        chat
    </button>
</form>
</body>
</html>

Code breakdown

This template does the following:

Displays a greeting title.
Shows a login form consisting of two fields:

Username (username) – a regular text field.
Room ID (room_id) – a field for entering a numeric identifier.

After clicking the "Enter chat" button, the data is sent to the server via POST (/join_chat).

Why just enter the room ID?

In production projects, you could select a room from a list, but in our case it is easier to let the user enter the ID themselves. If the room exists, they will connect, if not, they will create a new one.

Chat page (index.html)

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>WebSocket Chat - Room {{ room_id }}</title>
    <script src="https://cdn.tailwindcss.com"></script>
</head>
<body class="bg-gray-100 flex flex-col items-center p-4">
<h1 class="text-2xl font-bold mb-4">WebSocket Chat - Room {{ room_id }}</h1>

<!-- Hidden element for storing room data -->
<div id="room-data" 
    data-room-id="{{ room_id }}" 
    data-username="{{ username }}" 
    data-user-id="{{ user_id }}" 
    class="hidden">
</div>

<!-- Message area -->
<div id="messages" 
    class="w-full max-w-lg h-96 overflow-y-auto border border-gray-300 bg-white p-4 rounded-lg shadow-md">
</div>

<!-- Input field and button -->
<div class="flex mt-4 w-full max-w-lg"> 
    <input id="messageInput" 
           type="text" 
           placeholder="Enter a message" 
           class="flex-1 p-2 border border-gray-300 rounded-l-lg focus:outline-none focus:ring-2 focus:ring-blue-500"/> 
    <button onclick="sendMessage()" 
            class="bg-blue-500 text-white px-4 py-2 rounded-r-lg hover:bg-blue-600">Submit 
    </button>
</div>
<script src="/static/index.js"></script>
</body>
</html>

Code breakdown

This template creates a chat interface:

Displays a title with the room number ({{ room_id }}).
Hidden block #room-data

Stores data sent by the server (room_id, username, user_id).
Made hidden with class="hidden".
In the future, JavaScript will extract this data.

Message area (#messages) - user messages will appear here.
Input field and "Send" button

Field (#messageInput) for entering text.
The onclick="sendMessage()" button sends a message.

Connecting index.js

The /static/index.js file will describe the logic of interaction with WebSocket.

Why the hidden #room-data block?

Passing server variables directly to JS code is not always convenient. With Jinja2, you can output them to HTML and then read them in JavaScript.

2. Including styles

We use TailwindCSS, which is included via CDN:

<script src="https://cdn.tailwindcss.com"></script>

This allows us to write compact and flexible CSS code directly in HTML, for example:

<body class="bg-gray-100 flex flex-col items-center p-4">

Why TailwindCSS?

Eliminates the need to write custom CSS files.
Allows you to quickly style elements.
Good for prototyping.

3. How pages interact with the server

The user goes to the main page of the site (/).
Enters the name and room ID.
Clicks "Join the chat" - the request is sent to /join_chat.
The server returns index.html with the parameters passed to it.
The page loads, and the JS code starts working with WebSocket.

What's next?

We have prepared the interface, but the chat cannot send messages yet. In the next section, we will implement JavaScript logic that will allow us to connect to WebSocket, send and receive messages.

Implementing WebSocket in JavaScript

Now that we have an interface, it's time to add logic for exchanging messages in real time. We will use WebSocket for communication between the client and the server.

Implementing WebSocket in JavaScript

Now that we have the interface, it's time to add the logic for exchanging real-time messages. We will use WebSocket for communication between the client and the server.

A small but important digression

I want you to have a correct understanding: implementing WebSocket, and especially chats, on the backend side is not such a difficult task.

For us, backend developers, it is enough to write a few dozen lines of code to create a WebSocket connection, handle client connections and forward messages to all participants of a chat, room or, for example, an online game.

But if anyone has to really sweat, it is the frontend developers.

Why?

In addition to the visual component of the interface, they need to establish a connection to the server correctly.
Handle WebSocket events: connection, disconnection, receiving and sending messages.
Organize convenient and dynamic display of messages in real time.

Of course, this work is not super complicated, but the main difficulties when working with WebSocket often arise on the frontend.

Next, I will show a simple example of how to implement a WebSocket connection in JavaScript and how to work with it on the client side.

We describe the further code in the file app/static/index.js.

1. WebSocket initialization

First, we need to get the user and room data. Since we passed it via Jinja2 in a hidden HTML block (#room-data), we will extract it using JavaScript:

// Get data from a hidden element
const roomData = document.getElementById("room-data");
const roomId = roomData.getAttribute("data-room-id");
const username = roomData.getAttribute("data-username");
const userId = roomData.getAttribute("data-user-id");

We will need this data to establish a WebSocket connection.

2. Setting up a WebSocket connection

Let's create a connection to the server:

const ws = new WebSocket(`ws://localhost:8000/ws/chat/${roomId}/${userId}?username=${username}`);

At the deployment stage to the Amverum Cloud service, we will need to replace this link with a live https domain.

What's going on here?

We create a WebSocket connection at ws://localhost:8000/ws/chat/.
We pass the room ID (roomId), user ID (userId), and username (username) to the URL.
The server uses this data to identify the connection.

Additionally, we add event handlers to track the connection state:

ws.onopen = () => {
    console.log("Connection established");
};

ws.onclose = () => {
    console.log("Connection closed");
};

3. Receiving messages

When the server sends a message, the onmessage handler is triggered. We parse the JSON data and add it to the chat area:

ws.onmessage = (event) => {
    const messages = document.getElementById("messages");
    const messageData = JSON.parse(event.data);
    const message = document.createElement("div");

    // Define styles depending on the sender
    if (messageData.is_self) {
        message.className = "p-2 my-1 bg-blue-500 text-white rounded-md self-end max-w-xs ml-auto";
    } else {
        message.className = "p-2 my-1 bg-gray-200 text-black rounded-md self-start max-w-xs";
    }

    message.textContent = messageData.text;
    messages.appendChild(message);
    messages.scrollTop = messages.scrollHeight; // Auto scroll down
};

Code parsing:

JSON.parse(event.data) – converts a JSON string to an object.
If is_self == true, then the message was sent by the current user, and it is displayed on the right (in blue).
Otherwise, the message is from another user, and it is displayed on the left (in gray).
Autoscroll (messages.scrollTop = messages.scrollHeight) – so that new messages are always visible.

4. Sending messages

Let's create a function for sending messages:

function sendMessage() {
    const input = document.getElementById("messageInput");
    if (input.value.trim()) {
        ws.send(input.value);
        input.value = '';
    }
}

How does it work?

Take the text from the input field (#messageInput).
If the message is not empty, send it via ws.send().
Clear the input field.

Additionally, to send messages by Enter, add an event handler:

document.getElementById("messageInput").addEventListener("keypress", (e) => {
    if (e.key === "Enter") {
        sendMessage();
    }
});

Now the user can press Enter and the message will be sent automatically.

5. Full chat cycle

Connect to the WebSocket server when the page loads.
Wait for incoming messages and display them in the chat window.
When sending a message, the user enters the text and presses the button or Enter.
The server transmits the message to all participants in the room.
The chat is updated in real time without reloading the page.

Thanks to WebSocket and a small amount of JavaScript code, we have a full-fledged chat that works in real time.

Now we just need to configure the main file (the launch file) and we can start testing our web application.

Final settings and launching the application

At this stage, we only need to make the final settings and launch our application.

We will work with the file app/main.py, where we will implement three key tasks:

Initialize the FastAPI application
Add processing of static files (for example, JS, CSS)
Register all routes

Setting up main.py

Add the necessary imports and describe the basic configuration:

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from app.api.router_page import router as router_page
from app.api.router_socket import router as router_socket

app = FastAPI()

# Connect the folder with static files
app.mount('/static', StaticFiles(directory='app/static'), 'static')

# Register routes
app.include_router(router_socket)
app.include_router(router_page)

Since our web application is responsible for rendering HTML pages itself, we do not need CORS settings.

Launching the application

To start, we use uvicorn, a lightweight ASGI server that we installed earlier.

Basic command:

uvicorn main:app

By default, the server will go up to http://127.0.0.1:8000.

If you need to configure the launch parameters, add flags:

--host 0.0.0.0 – makes the server accessible from the external network
--port 8005 – specifies a specific port
--reload – enables auto-restart when changing the code (useful in development)

Example: Launch on port 8005 with auto-update of the code

uvicorn main:app --port 8005 --reload

After a successful launch, the server is ready to work, and we can connect to the chat! 🎉

Below is a short screencast demonstrating the application in action (3 users in one room):

This is all great, but chatting with yourself is not that exciting. It would be great to make this chat available to everyone. And the Amverum Cloud service will help us with this.

Why Amverum?

I chose this service because it is easy to deploy applications. You can see for yourself. To deploy an application similar to ours, you only need to follow a few simple steps:

Register on the service.
Click on the "Create project" button.
Upload the project files to the service. This can be done either through the interface on the site or using Git commands.
Fill in the necessary configurations on the site (for example, select a programming language for the project or specify a command to launch it).
Attach a free https domain name to the project or register your own.

The deployment process takes only a couple of minutes. In addition, each new user receives a bonus of 1$ to the main balance. The choice of the service is obvious.

Let's perform the deployment.

Register on the Amverum Cloud website, if you haven't registered yet
Click on "Applications"
Then select "Create an application" and follow the step-by-step instructions.

Name the project and choose a tariff plan (a Tester plan is suitable for educational purposes)

Upload files in a convenient way method

Fill in the configurations. The command to run will be: uvicorn app.main:app --host 0.0.0.0 --port 8000

Next, to activate the free https domain, we need to go to the created project, then go to the "Domains" tab and activate the free domain, as shown below:

Now we need to make some changes to the static/index.js file. In particular, we must replace the web socket connection string.

Before:

const ws = new WebSocket(`ws://localhost:8000/ws/chat/${roomId}/${userId}?username=${username}`);

After:

const ws = new WebSocket(`wss://easysocketfastap-yakvenalex.amverum.com/ws/chat/${roomId}/${userId}username=${username}`);

Note that we not only replaced the link to the domain name, but also corrected the format from ws to wss. Be careful!

Now all that remains is to rewrite the index.js file in Amverum and rebuild the project.

You can see the working code here: https://easysocketfastap-yakvenalex.amverum.com. The full source code of the project, as well as other exclusive content that I do not publish on Habr, can be found in my free telegram channel "Easy Path to Python".

By the way, I analyzed the project we talked about today in my half-hour video, available on YouTube and RuTube.

Conclusion

Today's material is a methodological basis that will prepare you for analyzing a more complex and large-scale project: a Telegram bot with MiniApp for anonymous chat "Tete-a-Tet". In the next article, we will analyze this project in detail, and the knowledge gained here will help you master it easier.

I tried to make the material as accessible and understandable as possible, so that even complex topics became simpler. If you found this article useful, don't forget to support it with a like or a comment - it motivates me to prepare even more interesting content.

And if you want even more practice and useful analysis, subscribe to my Telegram channel "Easy Path to Python" - there are already almost 3000 participants, and we regularly discuss interesting topics on development.

That's all for now. See you in the next materials!

Assistente Pessoal de IA nos Seus Dados. Parte 1: Vector ChromaDB + DeepSeek | GPT

Amverum Cloud — Tue, 06 May 2025 05:47:39 +0000

Olá a todos! Hoje gostaria de abordar um tema que interessa a muitos: a conexão de um modelo de linguagem grande como DeepSeek ou ChatGPT com sua própria base de conhecimento.

Neste artigo, vou explicar detalhadamente os princípios dos bancos de dados vetoriais e por que eles podem ser usados como parte da conexão da sua base de conhecimento com redes neurais “grandes” já prontas.

Como exemplo, vamos considerar a busca na documentação da Amverum Cloud. Amverum Cloud é uma alternativa ao Heroku mais barata e fácil de implantar, com domínios gratuitos, armazenamento persistente incluído e a possibilidade de atualizar projetos via git push amverum master.

Atualmente estamos desenvolvendo ativamente um agente de IA que ajudará os usuários a implantar projetos na nuvem, eliminar erros no código e na configuração, e também simplificará o trabalho com a documentação. E a busca na documentação é um excelente caso para usar um banco de dados vetorial, como o Chroma.

Vamos analisar imediatamente um bloco limitante

Redes neurais já prontas, como DeepSeek, ChatGPT ou Claude, são projetadas inicialmente para receber certos dados de texto com um contexto definido (prompt) como entrada e, com base em seus parâmetros e treinamento, processam sua solicitação, realizando a tarefa necessária.

Ou seja, como trabalharemos com modelos de linguagem já prontos, não poderemos nos desviar do processo de preparar prompts com um contexto definido e enviar solicitações para as redes neurais já prontas.

Isso leva a uma tarefa lógica: como conectar nosso banco de dados ao processo de comunicação com a rede neural? Imagine que sejam milhares de páginas de literatura profissional. É duvidoso que formemos centenas de megabytes de dados em um único prompt gigante e, ao mesmo tempo, esperemos que a rede neural aceite trabalhar com isso. No mínimo, receberemos um erro de que o contexto transmitido foi excedido.

É lógico que a solução sugere-se por si só: formar sua solicitação, complementando-a apenas com o contexto necessário para essa solicitação específica, mas como fazer isso? A opção mais simples é organizar algum tipo de motor de busca SEO. Enviamos, por exemplo, “treinamento de redes neurais” e é feita uma busca em nossos gigabytes de dados, mas aqui surge um problema lógico. Mesmo que encontremos informação, digamos, mil ocorrências da frase “treinamento de redes neurais”, como destacamos esse contexto? Como decidimos o que é importante transmitir à rede neural para a comunicação e o que não é necessário? Claro, você pode complicar muito esse motor de busca, mas, como entende, essa abordagem não é ótima, e isso significa que é preciso buscar outra solução mais flexível e bem-sucedida, sobre a qual falaremos hoje em detalhes.

Bancos de dados vetoriais

Provavelmente você nunca ouviu falar desses bancos de dados, ou só conhece superficialmente. Por isso, a seguir vou explicar detalhadamente e de forma simples o que são esses bancos de dados e, o mais importante, entenderemos como eles nos ajudarão em nossa tarefa.

Resumindo e de forma simples, um banco de dados vetorial é uma representação de dados: texto ou bytes em uma representação numérica ou, mais corretamente, em forma de vetores.

Por exemplo, a palavra “olá” em representação vetorial será assim: [-0.012, 0.124, -0.056, 0.203, …, 0.078] (removi números desnecessários para maior clareza).

No contexto atual, não importa muito como esses números são obtidos matematicamente. Agora só é importante uma compreensão geral da representação digital dos dados.

Benefício prático

Neste ponto, você pode dizer com razão: “Tudo isso é divertido, mas qual o uso prático de tudo isso e como se obtêm esses números estranhos?” Vamos descobrir.

A primeira coisa a entender é que as redes neurais não podem ler seus textos nem ver suas fotos no sentido humano. Por exemplo, quando você envia uma solicitação ao ChatGPT, acontece o seguinte:

A rede neural recebe o texto.
Transforma esse texto em vetores ou, mais corretamente, em embeddings.
Usa esses embeddings para gerar uma resposta baseada nos parâmetros treinados do modelo.
Realiza os cálculos.
Recebe uma resposta em forma de embedding.
Transforma o embedding em texto humano (Linguagem Natural).
Envia a você o texto humano.

Ou seja, já nesta etapa fica claro que na comunicação com redes neurais aparece algo como uma representação numérica da informação recebida (representação em forma de embeddings), e agora vamos descobrir por que isso é necessário.

Embeddings e para que servem

Os embeddings são “impressões digitais do significado” de um texto. Vamos ver este conceito com um exemplo real.

Imagine que você tem uma pilha enorme de documentos: livros, artigos, anotações. E quer encontrar informações neles não por palavras exatas, mas por significado.

Como implementar isso:

A rede neural aceita todas as palavras que você preparou e as converte em sequências de números (embeddings).

Exemplo:

“Gato” → [0.2, -0.5, 0.7, …]
“Gatinho” → 0.19, -0.48, 0.69, …

Um banco de dados vetorial, como o Chroma, armazena esses números e pode:

Buscar vetores que estejam próximos em significado (mesmo que as palavras sejam diferentes).
Responder a perguntas como: “Encontre algo sobre animais de estimação com bigodes” → vai retornar documentos sobre gatos, gatinhos.

Analogia simples

Busca normal (como Ctrl+F): busca por coincidências exatas de palavras.
A consulta “carro” não encontrará “automóvel”.
Busca por embeddings: busca por significados semelhantes.
“Carro”, “automóvel”, “transporte” - estarão próximos nos números e o banco de dados os relacionará.

Por que isso é necessário?

Os chatbots (como o ChatGPT) usam embeddings para entender consultas por significado, não por coincidências exatas.

Você pode perguntar a um documento com suas próprias palavras, e o sistema entenderá, mesmo que não haja coincidências exatas.

Resultado: Os embeddings são um “tradutor” de texto para números, para que você possa buscar por significado, não por letras. Os bancos de dados vetoriais são seu “armazenamento”, onde a busca funciona como um ímã para ideias semelhantes.

Existem redes neurais especiais que podem transformar seus dados de texto em uma representação vetorial (numérica). Para tais tarefas, não é necessário recorrer a gigantes como ChatGPT ou DeepSeek, basta usar redes neurais especializadas. Mais adiante, na parte prática, mostrarei a criação de embeddings e a implementação de busca inteligente usando como exemplo a rede neural paraphrase-multilingual-MiniLM-L12-v2.

Essa rede neural pesa apenas 500 MB e roda localmente até mesmo em computadores medianos, por isso é ideal para nossas tarefas.

Bancos de dados vetoriais (continuação)

Agora, você já deve ter entendido que, ao trabalhar com dados em representação numérica, aplica-se a seguinte conexão:

Os dados de texto são transformados em uma representação vetorial (numérica) por meio de uma rede neural especial.
Para que a busca inteligente funcione, precisamos:
- Obter o embedding do seu conjunto de dados em representação vetorial.
- Escrever uma consulta de texto como “encontre tudo o que se carrega na mão”.
- Transformar a consulta em representação vetorial.
- Comparar o vetor da consulta com os vetores do grande embedding.
- Transformar o resultado encontrado de volta para linguagem humana.
- Fornecer o resultado.

Agora surge uma pergunta lógica sobre o armazenamento desses dados numéricos. É claro que podemos pegar o texto, transformá-lo em vetores usando alguma, por enquanto desconhecida, paraphrase-multilingual-MiniLM-L12-v2, mas onde armazenar isso?

Isso pode ser armazenado em um banco de dados especializado, como o Chroma, com o qual trabalharemos hoje, ou na RAM.

Princípio do armazenamento em RAM

Preparar os dados de texto. Transformá-los em embeddings.

Enviar uma solicitação de embedding.

Depois que o script termina de ser executado, a memória é limpa e, ao iniciar novamente, é necessário gerar os embeddings novamente.

Princípio do armazenamento em um banco de dados especializado

Preparar os dados de texto.

Transformá-los em embeddings.

Salvar o embedding no formato do banco de dados.

Ao trabalhar, conectar-se ao banco de dados e usar o embedding já pronto.

Os bancos de dados vetoriais podem ser tanto locais (self-hosted) quanto em nuvem.

O banco de dados local mais simples é o ChromaDB, com o qual trabalharemos hoje. Escolhi por sua simplicidade e ótima integração com a linguagem Python.

Outros bancos de dados vetoriais locais: Qdrant, Weaviate, Milvus e outros.

Aqui alguns exemplos de bancos de dados em nuvem: Pinecone, RedisVL, Qdrant Cloud e outros.

O que vamos fazer hoje?

Mais adiante haverá ainda mais teoria, mas tenho certeza de que você está esperando a parte prática. Então, vamos descobrir o que faremos na prática, além de absorver a informação teórica.

A seguir, mostrarei com um exemplo prático como transformar dados de texto em um banco de dados vetorial ChromaDB. Primeiro, criaremos um banco de dados usando um exemplo simples e trabalharemos com a busca nele, e depois trabalharemos com um exemplo mais complexo, sobre o qual construiremos nosso assistente inteligente no futuro.

Depois de entender completamente como funciona a geração de embeddings e a busca inteligente dentro deles, conectaremos as grandes redes neurais DeepSeek e GPT ao contexto geral, implementando a seguinte lógica:

O usuário envia uma solicitação específica.
Usando a busca inteligente, extraímos conteúdo adicional do nosso banco de dados vetorial.
Reenviamos a solicitação do usuário junto com nossos “resultados da busca inteligente” para DeepSeek / GPT e recebemos uma resposta da rede neural baseada no nosso contexto pessoal.

Vai ser interessante!

Banco de dados vetorial simples

Agora, para consolidar o bloco teórico geral descrito acima, prepararemos um banco de dados ChromaDB simples baseado na descrição de produtos de uma loja online fictícia (para o exemplo mais simples, é um exemplo visual).

Imagine que temos um conjunto de produtos deste tipo:

SHOP_DATA = [
    {
        "text": 'Portátil Lenovo IdeaPad 5: 16 Gb RAM, SSD 512 Gb, LED 15.6", precio 550 $.',
        "metadata": {
            "id": "1",
            "type": "product",
            "category": "laptops",
            "price": 55000,
            "stock": 3,
        },
    },
    {
        "text": 'Smartphone Xiaomi Redmi Note 12: 128 Gb RAM, SSD 512 Gb, LED 15.6", precio 1550 $.',
        "metadata": {
            "id": "2",
            "type": "product",
            "category": "phones",
            "price": 18000,
            "stock": 10,
        },
    },
    # ...
]

Como pode ver, a informação é apresentada como uma lista de dicionários, cada um contendo duas chaves principais: texto e metadados.

Documentos e metadados em bancos de dados vetoriais

Este é um ponto importante que vale a pena analisar com mais detalhe. Deve compreender que na base de dados ChromaDB, toda a nossa pesquisa “inteligente” funcionará “prontamente” apenas no valor da chave de texto (o nome pode ser qualquer coisa).

No contexto dos bancos de dados vetoriais, distinguem-se conceitos como documentos (no nosso caso, é o texto que descreve um produto específico) e metadados.

Restrições de comprimento de documento

Na etapa de preparação dos dados de texto para carregá-los em um banco de dados vetorial, é extremamente importante levar em conta as restrições de comprimento de texto estabelecidas pelo modelo que irá convertê-los para uma representação vetorial. Essas restrições dependem da rede neural específica e geralmente variam de 256 a 512 tokens.

Portanto, se, por exemplo, você planeja adicionar um livro inteiro de 200 páginas ao banco de dados, primeiro ele deve ser dividido em trechos menores, cada um não maior que 256–512 tokens. Isso garantirá que o modelo processe corretamente o texto e preserve a integridade semântica no espaço vetorial.

O modelo paraphrase-multilingual-MiniLM-L12-v2 com o qual trabalharemos hoje tem um limite de 512 tokens por documento. Isso equivale a aproximadamente 250–350 palavras ou 1500–2200 caracteres. No entanto, como usaremos nossos dados com as poderosas redes neurais DeepSeek e ChatGPT, faremos o tamanho do documento ainda menor para evitar sobrecarregar o contexto.

O papel dos metadados

Os metadados são nosso assistente, tornando a busca mais precisa e localizada. Aqui está um exemplo simples:

Imagine que sua loja online vende notebooks, smartphones e outros equipamentos. Você pode limitar a busca inteligente pela categoria dos metadados. Independentemente de você usar os metadados para a busca explicitamente, é importante entender que os documentos encontrados (resultados da busca) sempre conterão essas tags meta, o que permitirá processamento adicional.

Exemplo com livros:

Imagine que você decidiu carregar toda a série de livros de Harry Potter em um banco de dados vetorial. Você dividiu todo o texto em trechos de 512 tokens e especificou as seguintes tags meta: título do livro, número da página, número do capítulo e outras informações.

Depois, você faz uma consulta como “Quando Harry Potter soube sobre as Horcruxes”. Sua busca inteligente retorna certos resultados: documentos com tags meta que permitem entender de qual livro vem e em que página está a informação.

Outro exemplo de trabalho com tags meta é limitar a busca a um livro específico, série de livros ou autor ao formar uma consulta.

Lembre-se de que a busca inteligente não está relacionada aos metadados de forma alguma, a menos que você especifique explicitamente a lógica para processá-los na consulta.

Uso prático dos metadados

No contexto dos metadados para nossos produtos:

{
    "id": "1",
    "type": "product",
    "category": "laptops",
    "price": 55000,
    "stock": 3,
}

Transmitimos o ID do produto, tipo, categoria, preço e estoque. Por exemplo, uma busca inteligente de produtos poderia ser implementada no site e apenas os produtos com o ID encontrado seriam exibidos ao usuário na página de resultados.

Assim, um banco de dados vetorial é uma excelente ferramenta não apenas para integração com redes neurais, mas também para resolver tarefas práticas, como um motor de busca inteligente em um site ou em sua API.

Criando um Banco de Dados Vetorial em Python

Agora vamos escrever algum código. O código será simples, uma vez que não precisamos de dividir documentos (descrições de produtos) em fragmentos ou executar lógica adicional para analisar e processar produtos.

Preparar o projeto

A primeira coisa que fazemos é criar um novo projeto Python com um ambiente virtual dedicado. Adicione um ficheiro requirements.txt ao projeto com o seguinte conteúdo:

langchain-huggingface==0.1.2
torch==2.6.0
loguru==0.7.3
chromadb==0.6.3
sentence-transformers==3.4.1
langchain-chroma==0.2.2
pydantic-settings==2.8.1
langchain-text-splitters==0.3.7
langchain-deepseek==0.1.3
langchain-openai==0.3.11

O que é o langchain?

Já deve ter reparado que encontrei a palavra langchain nas dependências até 5 vezes, o que não é acidental.

Langchain é uma poderosa ferramenta Python que lhe permite integrar um grande número de redes neuronais e outras ferramentas úteis ao seu projeto, prontas a utilizar. Este framework merece ser mencionado em mais do que um artigo, pelo que recomendo vivamente que leia a documentação do projeto (mais de 105 mil estrelas no GitHub devem indicar que o projeto merece atenção).

No projeto de hoje, iremos utilizar as seguintes ferramentas deste framework:

langchain-text-splitters==0.3.7

Uma ferramenta para dividir textos grandes em fragmentos mais pequenos (pedaços). Contém várias estratégias de segmentação: por símbolos, frases, tokens, blocos semânticos. É necessário para processar documentos grandes quando estes excedem a janela de contexto dos modelos de linguagem (256–512 tokens no nosso caso).

langchain-deepseek==0.1.3

Integração com os modelos de IA da DeepSeek. Permite utilizar modelos de linguagem DeepSeek poderosos para análise e geração de texto (hoje a demonstração será baseada no modelo deepseek-chat).

langchain-openai==0.3.11

Adaptador para trabalhar com a API OpenAI. Fornece acesso à família de modelos de linguagem GPT e às suas funções. Inclui suporte para modelos de chat, modelos de incorporação e funções para trabalhar com imagens via DALL-E (vamos considerar um exemplo com o modelo gpt-3.5-turbo).

langchain-chroma==0.2.2

Conector para a base de dados vetorial ChromaDB. Permite armazenar e recuperar eficientemente representações vetoriais de textos para implementar funções de pesquisa semântica e construir bases de conhecimento com a capacidade de pesquisar por similaridade.

langchain-huggingface==0.1.2

A integração com o ecossistema Hugging Face dá acesso a milhares de modelos abertos desenvolvidos pela comunidade. Isto inclui modelos de linguagem, modelos para incorporações, classificação e outras tarefas de PNL. Esta ferramenta é especialmente útil quando precisa de executar modelos localmente ou utilizar modelos especializados. No nosso caso, poderemos utilizar a rede neural local paraphrase-multilingual-MiniLM-L12-v2.

Outros pacotes

torch: Uma biblioteca de aprendizagem automática que alimenta muitos modelos de linguagem
loguru: Um registador prático para acompanhar o progresso do nosso código
chromadb: A base de dados vetorial real que usaremos
sentence-transformers: Uma biblioteca para converter texto em vetores (embeddings)

Instalando dependências

Agora vamos instalar:

pip install -r requirements.txt

A instalação pode demorar muito tempo devido ao peso total dos pacotes que estão a ser instalados (no meu caso, o processo de instalação demorou cerca de 20 minutos).

Criando um banco de dados

Vamos agora criar um ficheiro chamado create_chromadb.py (pode dar-lhe qualquer nome).

Realizar importações:

import time
from langchain_huggingface import HuggingFaceEmbeddings
import torch
from loguru import logger
from langchain_chroma import Chroma

Vamos definir as principais variáveis:

CHROMA_PATH = "./shop_chroma_db"
COLLECTION_NAME = "shop_data"

As coleções em bases de dados vetoriais podem ser comparadas a tabelas em bases de dados clássicas. Se não especificar um nome de coleção, o ChromaDB fornece um valor predefinido.

Vamos agora adicionar uma variedade de produtos (quanto mais, melhor):

SHOP_DATA = [
    {
        "text": 'Lenovo IdeaPad 5: 16 RAM, SSD 512, 15.6", prise 55000 $.',
        "metadata": {
            "id": "1",
            "type": "product",
            "category": "laptops",
            "price": 55000,
            "stock": 3,
        },
    },
    # add other products
]

E por fim, vamos escrever uma função para criar a base de dados:

def generate_chroma_db():
    try:
        start_time = time.time()
                logger.info("Carregando o modelo de incorporação...")
        embeddings = HuggingFaceEmbeddings(
            model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
            model_kwargs={"device": "cuda" if torch.cuda.is_available() else "cpu"},
            encode_kwargs={"normalize_embeddings": True},
        )
        logger.info(f"Modelo carregado em {time.time() - start_time:.2f} seg")
                logger.info("Criação do Chroma DB...")
        chroma_db = Chroma.from_texts(
            texts=[item["text"] for item in SHOP_DATA],
            embedding=embeddings,
            ids=[str(item["metadata"]["id"]) for item in SHOP_DATA],
            metadatas=[item["metadata"] for item in SHOP_DATA],
            persist_directory=CHROMA_PATH,
            collection_name=COLLECTION_NAME,
        )
        logger.info(f"Chroma DB criado em {time.time() - start_time:.2f} seg")
                return chroma_db
    except Exception as e:
        logger.error(f"Erro: {e}")
        raise

Neste código nós:

Crie um modelo para converter texto em vetores (embeddings)
Inicialize a base de dados Chroma com os nossos produtos
Transferir textos, IDs, metadados e definições do espaço de pesquisa
Registar criação bem-sucedida de base de dados ou erros

É importante referir que utilizámos o modelo multilingue paraphrase-multilingual-MiniLM-L12-v2, que funciona bem com a língua portuguesa. Isto permitir-nos-á criar corretamente representações vetoriais para os nossos produtos em russo. O modelo em si será carregado automaticamente a partir do repositório Hugging Face na primeira execução. Não haverá necessidade de o carregar em execuções repetidas se o modelo já estiver na sua máquina.

A particularidade do modelo escolhido é que é equilibrado em tamanho e qualidade — oferece bons resultados de vetorização de texto mesmo no CPU, mas com uma placa gráfica compatível com CUDA funciona muito mais rapidamente. Além disso, graças ao parâmetro normalize_embeddings=True, todos os embeddings criados serão normalizados, o que aumentará a precisão da pesquisa de produtos semelhantes.

Execute o código:

if __name__ == "__main__":
   generate_chroma_db()

Feito! Temos agora uma base para criar uma base de dados vetorial com uma descrição de documentos. Na raiz do projeto após o lançamento, deverá aparecer uma pasta chamada “shop_chroma_db”, na qual estarão localizados os ficheiros da base de dados. No futuro, poderemos utilizar esta base de dados para a pesquisa semântica de produtos por solicitação do utilizador — a pesquisa encontrará não só correspondências exatas, mas também produtos semanticamente próximos.

Escrevendo um mecanismo de busca para o banco de dados

Agora que a base de dados está pronta, podemos descrever um sistema de pesquisa simples dentro da mesma. O princípio aqui será o seguinte:

Ligar a uma base de dados existente, recebendo embeddings
Receba uma consulta de pesquisa do utilizador
Transforme a consulta de pesquisa numa representação vetorial
Comparar o vetor de consulta recebido com os dados da base de dados
Devolver ao utilizador o número de documentos que este solicitou

Quanto à emissão de documentos ao utilizador. Há aqui uma característica: indicamos sempre o número de documentos que queremos receber. Digamos que podem ser 5 peças. Se não ligar estritamente os metadados através de filtros, receberá sempre como resposta da base de dados exatamente o número de documentos especificado.

Ou seja, mesmo que a saída não corresponda de todo ao seu pedido, continuará a receber os seus 5 documentos como resposta. Isto é importante considerar ao desenvolver uma interface de utilizador.

A resposta em si consiste em: um documento, metadados ligados a este e um índice de classificação (quanto mais baixo for o índice, mais semelhante em significado será o documento que irá receber). A emissão em si baseia-se no princípio: quanto mais baixo for o índice, maior será a posição na emissão.

Agora vamos voltar ao código.

Importações:

from langchain_huggingface import HuggingFaceEmbeddings
import torch
from loguru import logger
from langchain_chroma import Chroma

Variáveis:

CHROMA_PATH = "./shop_chroma_db"
COLLECTION_NAME = "shop_data"

Vamos agora descrever a função para ligar a uma base de dados existente:

def connect_to_chroma():
    """Ligue-se ao Chroma."""
    try:
        logger.info("Incorporação de download...")
        embeddings = HuggingFaceEmbeddings(
            model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
            model_kwargs={"device": "cuda" if torch.cuda.is_available() else "cpu"},
            encode_kwargs={"normalize_embeddings": True},
        )
        chroma_db = Chroma(
            persist_directory=CHROMA_PATH,
            embedding_function=embeddings,
            collection_name=COLLECTION_NAME,
        )
        logger.success("Sucesso na ligação com o Chroma")
        return chroma_db
    except Exception as e:
        logger.error(f"Erro ao ligar o Chroma: {e}")
        raise

Esta função irá retornar um objeto de base de dados pronto a usar, dentro do qual a pesquisa vetorial será realizada.

Vamos agora descrever uma função que permitirá a pesquisa tanto por documento (páginas de documentação) como por metadados:

def search_products(query: str, metadata_filter: dict = None, k: int = 4):
    """
    Pesquisa de página
        """
    try:
        chroma_db = connect_to_chroma()
        results = chroma_db.similarity_search_with_score(
            query, k=k, filter=metadata_filter
        )
        logger.info(f"Encontrado {len(resultados)} resultado da consulta: {query}")
        formatted_results = []
        for doc, score in results:
            formatted_results.append(
                {
                    "text": doc.page_content,
                    "metadata": doc.metadata,
                    "similarity_score": score,
                }
            )
        return formatted_results
    except Exception as e:
        logger.error(f"Erro: {e}")
        raise

E agora vamos chamar-lhe:

for i in search_products(query="como implantar uma aplicação no Amverum, com git push?"):
    print(i)

O parâmetro metadata_filter permite-nos especificar condições de filtragem de metadados. O exemplo de pesquisa de produtos de loja online é mais adequado aqui do que a nossa documentação. Por exemplo, se quisermos encontrar apenas portáteis com um preço abaixo dos 600 dólares, podemos usar um filtro como este:

filter_condition = {
    "category": "laptops",
    "price": {"$lt": 600}
}

Aqui quero salientar que este exemplo é educativo. Num sistema de produção, é normalmente criada uma classe para gerir uma sessão com uma base de dados, para que se possa ligar uma vez e manter a ligação. Na implementação atual, cada chamada à função irá reiniciar a ligação e carregar o modelo de incorporação, o que consome muitos recursos. No próximo artigo, se perceber que está interessado neste tema, vou mostrar-lhe como implementar esta classe e como pode criar uma aplicação web completa baseada numa base de dados vetorial.

A propósito, o código-fonte do projeto de hoje, bem como a classe já escrita para gerir a ligação à base de dados vetorial, podem ser encontrados no meu canal de Telegram “Easy Way to Python”. O código-fonte completo do projeto está lá há cerca de uma semana.

Ligue para a pesquisa e veja os resultados.

Tenha em atenção o valor da pontuação de similaridade. Quanto mais baixo for este valor, mais próximo estamos do significado da consulta.

Nada mau, certo? Embora não tenhamos especificado explicitamente um modelo específico de aspirador ou as suas características na consulta, a pesquisa vetorial foi capaz de compreender a semântica da consulta e devolver os resultados mais relevantes da nossa base de dados. Isto demonstra o poder das bases de dados vetoriais combinadas com modelos de linguagem modernos.

Vamos praticar

Agora que já tem uma compreensão básica dos princípios de trabalho com bases de dados vetoriais, podemos começar a criar uma lógica mais complexa.

Falaremos do serviço Amverum Cloud, que já referi nos meus artigos anteriores sobre o Habr. Em suma, é uma plataforma para a fácil implementação (lançamento remoto) de projetos em quase qualquer linguagem de programação.

Uma das vantagens do Amverum Cloud é a capacidade de implementar rapidamente um projeto em diversas tecnologias utilizando o git push (ou arrastando ficheiros na interface), bem como obter um nome de domínio gratuito para o mesmo. Além disso, o serviço oferece proxy integrado gratuito para as APIs OpenAI, Antropic, Cloude e Grok, o que é conveniente para interagir com o LLM. Graças a estas capacidades técnicas, o serviço conta com uma documentação muito extensa e detalhada — para todas as ocasiões: desde guias básicos, como a migração do Heroku, a tópicos mais específicos — por exemplo, lançamento de projetos em FastAPI ou Django, ou resolução de problemas de faturação.

Esta documentação (e são cerca de 70 documentos diferentes) é perfeita para uma demonstração prática: podemos pegar numa série de textos, transformá-la numa base de dados vectorial e mostrar como integrar eficazmente esta base de dados com modelos de redes neuronais para obter respostas significativas e úteis.

Dados de entrada

Já tinha em mãos cerca de 70 documentos em Markdown da Amverum neste formato:

# Suporte de sondas Kubernetes

## Como configurar?

O formulário requer o preenchimento das configurações em formato yaml, nativo do formato utilizado
pelo próprio k8s (ver [aqui](https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/)).
O ficheiro de configuração é substituído na sua implementação, que é carregada com ele no cluster.
**Isto significa que se carregou uma configuração que não funciona ou no formato errado, o seu projeto irá falhar na fase de compilação.** …

Estes documentos foram-me fornecidos como um projeto contendo ficheiros md e outros ficheiros técnicos. Mas muito provavelmente, no seu caso, não haverá esse presente.

Por conseguinte, é importante lembrar que para recolher informações para a base de dados, pode utilizar qualquer ferramenta conveniente para si: análise de sites, carregamento de dados das suas próprias bases de dados SQL, leitura de documentos nos formatos Excel, Word, PDF e assim por diante. O principal é que no final a informação seja convertida num formato suportado por bases de dados vectoriais e que tenha uma estrutura clara.

A primeira coisa que implementei no formato de uma tarefa com preparação foi transformar todos estes ficheiros com uma estrutura aninhada num formato JSON deste tipo:

{
   "text": "some text",
   "metadata": {
       "section_count": 2,
       "section_1": "backups",
       "section_2": "backups /data"
   }
}

Ou seja, são informações de texto integral na legenda do texto, sem sinais de pontuação, caracteres especiais e letras maiúsculas, e informações técnicas colocadas nos metadados, como o número de secções, nomes, etc.

É importante compreender que os metadados são os nossos assistentes, mas nada impede que carregue apenas textos. No entanto, com os metadados, obtemos oportunidades adicionais para filtrar e analisar os resultados de pesquisa.

Vamos descrever a lógica para criar uma pasta com ficheiros JSON:

import json
import os
import re
import string
import sys
from typing import Any, Dict, List
from config import settings
from loguru import logger

Aqui usei as bibliotecas internas do Python, com exceção do loguru — para um registo conveniente dos resultados. A extração em si será orientada para expressões regulares.

Vamos gerar as constantes:

HEADER_PATTERN = re.compile(r"^(#+)\s(.+)")
PUNCTUATION_PATTERN = re.compile(f"[{re.escape(string.punctuation)}]")
WHITESPACE_PATTERN = re.compile(r"\s+")

Vamos descrever a lógica para normalizar textos:

def normalize_text(text: str) -> str:
  """Normalização de texto: remoção de pontuação e caracteres especiais."""
  if not isinstance(text, str):
  raise ValueError("O texto de entrada deve ser uma string")
  # Removendo pontuação
  text = PUNCTUATION_PATTERN.sub(" ", text)
  # Removendo quebras de linha e espaços extra
  text = WHITESPACE_PATTERN.sub(" ", text)
  # Convertendo para minúsculas
  return text.lower().strip()

Vamos escrever a lógica para analisar um documento no formato md:

def parse_markdown(md_path: str) -> Dict[str, Any]:
    """Análise de dados"""
    if not os.path.exists(md_path):
        raise FileNotFoundError(f"Ficheiro {md_path} não encontrado")
    try:
        with open(md_path, "r", encoding="utf-8") as file:
            content = file.read()
    except Exception as e:
        logger.error(f"Erro de leitura {md_path}: {e}")
        raise
    sections: List[str] = []
    section_titles: List[str] = []
    current_section: str | None = None
    current_content: List[str] = []
    for line in content.splitlines():
        section_match = HEADER_PATTERN.match(line)
        if section_match:
            if current_section:
                sections.append("\n".join(current_content).strip())
                section_titles.append(current_section)
                current_content = []
            current_section = section_match.group(2)
            current_content.append(current_section)
        else:
            current_content.append(line)
    if current_section:
        sections.append("\n".join(current_content).strip())
        section_titles.append(current_section)
    # Normalização de texto
    normalized_sections = [normalize_text(section) for section in sections]
    full_text = " ".join(normalized_sections)
    # Estrutura de metadados
    metadata = {
        "file_name": os.path.basename(md_path),
        "section_count": len(section_titles),
    }
    # Adicionar cabeçalho
    for i, title in enumerate(section_titles):
        metadata[f"section_{i+1}"] = title
    return {"text": full_text, "metadata": metadata}

E descreveremos a função para analisar todos os documentos:

def process_all_markdown(input_folder: str, output_folder: str) -> None:
    """processamento de markdown"""
    if not os.path.exists(input_folder):
        raise FileNotFoundError(f"{input_folder} não encontrado")
    try:
        os.makedirs(output_folder, exist_ok=True)
    except Exception as e:
        logger.error(f"erro de criação do diretório de saída: {e}")
        raise
    for root, _, files in os.walk(input_folder):
        for file_name in files:
            if file_name.endswith(".md"):
                try:
                    md_path = os.path.join(root, file_name)
                    output_path = os.path.join(
                        output_folder, file_name.replace(".md", ".json")
                    )
                    parsed_data = parse_markdown(md_path)
                    with open(output_path, "w", encoding="utf-8") as file:
                        json.dump(parsed_data, file, ensure_ascii=False, indent=4)
                    logger.info(f"Resultar em poupança{output_path}")
                except Exception as e:
                    logger.error(f"Erro de processamento de ficheiros {file_name}: {e}")

A função recebe como entrada a pasta na qual se encontram todos os documentos e a pasta na qual serão gravados os ficheiros JSON recebidos.

A lógica de deteção de ficheiros md funciona de forma a pesquisar em todas as pastas aninhadas por diretórios com documentos. Isto é especialmente útil se a sua documentação tiver uma estrutura complexa de ficheiros e diretórios.

Agora chamamos:

if __name__ == "__main__":
    try:
        process_all_markdown(
            input_folder="caminho para a pasta com documentos",
            output_folder="caminho para a pasta para guardar JSON",
        )
    except Exception as e:
        logger.error(f"Erro crítico: {e}")
        sys.exit(1)

Vamos lançar.

Recebeu 63 documentos (um exemplo de todos os ficheiros está no código fonte do projeto)

Os ficheiros estão prontos! Agora só precisamos de ler todos os ficheiros e criar uma base de dados vetorial baseada neles.

Note que o nosso analisador extrai não só o texto do Markdown, mas também a estrutura do documento. Nos metadados, guardamos o nome do ficheiro e os títulos de todas as secções. Isto permite-nos não só pesquisar por conteúdo, mas também indicar com precisão de que secção da documentação foi retirado um fragmento específico, o que é muito útil para criar respostas significativas do assistente.

Criar uma base de dados

O processo de criação de uma base de dados para o Amverum não é muito diferente da abordagem anteriormente descrita, com a diferença de que iremos obter dados de ficheiros JSON existentes.

Antes de continuar, para um código mais estruturado, vamos criar um ficheiro . env na raiz do projeto e colocar nele as seguintes variáveis:

DEEPSEEK_API_KEY=sk-1234
OPENAI_API_KEY=sk-proj-1234

Ambas as chaves API devem estar aqui. Se não existirem chaves, não precisa de criar o ficheiro.

De seguida, crie o ficheiro config.py e preencha-o da seguinte forma:

import os
from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
class Config(BaseSettings):
    DEEPSEEK_API_KEY: SecretStr
    BASE_DIR: str = os.path.abspath(os.path.join(os.path.dirname(__file__)))
    DOCS_AMVERA_PATH: str = os.path.join(BASE_DIR, "amverum_data", "docs_amvera")
    PARSED_JSON_PATH: str = os.path.join(BASE_DIR, "amverum_data", "parsed_json")
    AMVERA_CHROMA_PATH: str = os.path.join(BASE_DIR, "amverum_data", "chroma_db")
    AMVERA_COLLECTION_NAME: str = "amvera_docs"
    MAX_CHUNK_SIZE: int = 512
    CHUNK_OVERLAP: int = 50
    LM_MODEL_NAME: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
    DEEPSEEK_MODEL_NAME: str = "deepseek-chat"
    OPENAI_MODEL_NAME: str = "gpt-3.5-turbo"
    OPENAI_API_KEY: SecretStr
    model_config = SettingsConfigDict(env_file=f"{BASE_DIR}/.env")
settings = Config()  # type: ignore

Nesta aula recolhemos todas as variáveis necessárias para o funcionamento do projeto. Como pode ver, os caminhos para as pastas importantes são aqui indicados (a sua estrutura pode ser diferente), os nomes das coleções, o tamanho máximo dos blocos e os nomes dos modelos de redes neuronais.

Preste especial atenção à declaração das chaves API:

OPENAI_API_KEY: SecretStr
DEEPSEEK_API_KEY: SecretStr

É importante descrevê-las não como strings normais, mas via SecretStr. Este é um requisito da biblioteca LangChain para o manuseamento seguro de dados confidenciais.

Agora podemos importar a variável de definições para qualquer lugar no código e aceder às definições necessárias através de um ponto.

Criação de uma base de dados

Vamos voltar à criação de uma base de dados. Comecemos pelas importações:

import json
import os
import sys
from typing import Any, Dict, List, Optional
import torch
from langchain_chroma import Chroma
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from loguru import logger
from config import settings

Aqui, entre outras coisas, vê a importação de RecursiveCharacterTextSplitter. Este componente permitir-nos-á dividir documentos de grandes dimensões em blocos semânticos (fragmentos de texto relacionados entre si por significado).

Vamos escrever uma função para carregar todos os ficheiros JSON existentes no formato de lista de dicionário:

def load_json_files(directory: str) -> List[Dict[str, Any]]:
    """Download JSON"""
    documents = []
    try:
        if not os.path.exists(directory):
            logger.error(f"O diretório {directory} não existe")
            return documents
        for filename in os.listdir(directory):
            if filename.endswith(".json"):
                file_path = os.path.join(directory, filename)
                try:
                    with open(file_path, "r", encoding="utf-8") as file:
                        data = json.load(file)
                        documents.append(
                            {"text": data["text"], "metadata": data["metadata"]}
                        )
                        logger.info(f"Ficheiro carregado: {filename}")
                except Exception as e:
                    logger.error(f"Erro ao ler o ficheiro {nome do ficheiro}: {e}")
        logger.success(f"Ficheiros JSON {len(documents)} carregados")
        return documents
    except Exception as e:
        logger.error(f"Erro ao carregar ficheiros JSON: {e}")
        return documents

Após a execução desta função, receberemos todos os dados dos ficheiros JSON como dicionários colocados na RAM. As informações destes dicionários podem ser utilizadas para criar um banco de dados vetorial.

Vamos agora descrever a lógica para dividir documentos grandes em fragmentos significativos:

def split_text_into_chunks(text: str, metadata: Dict[str, Any]) -> List[Any]:
    """Dividir o texto em pedaços, preservando os metadados."""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=settings.MAX_CHUNK_SIZE,
        chunk_overlap=settings.CHUNK_OVERLAP,
        length_function=len,
        is_separator_regex=False,
    )
    chunks = text_splitter.create_documents(texts=[text], metadatas=[metadata])
    return chunks

Como resultado desta função, temos ainda uma lista de dicionários, mas agora os textos grandes estão divididos em fragmentos semânticos. Ao mesmo tempo, cada um dos fragmentos contém os metadados necessários. Isto é importante para manter o contexto geral, mesmo quando se divide um texto grande em muitas partes.

Por fim, descreveremos a lógica para a criação de uma base de dados vetorial:

def generate_chroma_db() -> Optional[Chroma]:
    """Inicialização do ChromaDB com dados de ficheiros JSON."""
    try:
        # Crie um diretório para armazenar a base de dados se não existir
        os.makedirs(settings.AMVERA_CHROMA_PATH, exist_ok=True)
        # Carregando ficheiros JSON
        documents = load_json_files(settings.PARSED_JSON_PATH)
        if not documents:
            logger.warning("Não há documentos para adicionar à base de dados.")
            return None
        # Inicializar o modelo de incorporação
        embeddings = HuggingFaceEmbeddings(
            model_name=settings.LM_MODEL_NAME,
            model_kwargs={"device": "cuda" if torch.cuda.is_available() else "cpu"},
            encode_kwargs={"normalize_embeddings": True},
        )
        # Preparação de dados para o Chroma
        all_chunks = []
        for i, doc in enumerate(documents):
            chunks = split_text_into_chunks(doc["text"], doc["metadata"])
            all_chunks.extend(chunks)
            logger.info(
                f"Document {i+1}/{len(documents)} is divided into {len(chunks)} chunks"
            )
        # Criar um armazenamento vetorial
        texts = [chunk.page_content for chunk in all_chunks]
        metadatas = [chunk.metadata for chunk in all_chunks]
        ids = [f"doc_{i}" for i in range(len(all_chunks))]
        chroma_db = Chroma.from_texts(
            texts=texts,
            embedding=embeddings,
            ids=ids,
            metadatas=metadatas,
            persist_directory=settings.AMVERA_CHROMA_PATH,
            collection_name=settings.AMVERA_COLLECTION_NAME,
            collection_metadata={
                "hnsw:space": "cosine",
            },
        )
        logger.success(
            f"Baza Chroma inicializado, adicionado {len(all_chunks)} pedaços de {len(documents)} documentos"
        )
        return chroma_db
    except Exception as e:
        logger.error(f"Erro de arranque Chroma: {e}")
        raise

A lógica de criação de uma base de dados não é muito diferente dos exemplos que vimos anteriormente. A principal diferença é dividir os textos grandes em fragmentos significativos (pedaços).

No meu modesto computador, o processo de criação de uma base de dados demorou menos de 10 minutos. Em dispositivos com GPU, este processo demorará apenas alguns minutos graças à computação paralela.

Otimização de desempenho

De notar que ao trabalhar com grandes volumes de dados, pode otimizar ainda mais o processo:

Utilize lotes ao criar embeddings (processamento de documentos em grupos)
Configure os parâmetros MAX_CHUNK_SIZE e CHUNK_OVERLAP consoante a estrutura dos seus dados
Escolha um modelo de incorporação adequado, equilibrando a qualidade e a velocidade

O ChromaDB tem um bom desempenho para a maioria das tarefas RAG, mas para volumes de dados muito grandes (milhões de documentos), vale a pena considerar outras soluções, como o FAISS ou o Pinecone.

Integração de um motor de busca com redes neuronais

Anteriormente, vimos como descrever um motor de busca sem utilizar grandes redes neuronais. Para consolidar este material, sugiro que crie você mesmo um motor de busca com base na sua base de dados.

Passarei à parte mais interessante: explicarei como combinar a saída de uma base de dados vetorial com redes neuronais para criar um agente de IA inteligente. Para tal, vamos utilizar dados da documentação do Amverum Cloud, que já transformei numa base de dados vetorial.

Configuração inicial do projeto

Vamos começar por criar um ficheiro chamado chat_with_ai.py.

Vamos fazer as importações necessárias:

from typing import Any, Dict, List, Literal, Optional
import torch
from config import settings
from langchain_chroma import Chroma
from langchain_deepseek import ChatDeepSeek
from langchain_openai import ChatOpenAI
from langchain_huggingface import HuggingFaceEmbeddings
from loguru import logger

Aqui incluímos as novas ferramentas langchain_deepseek e langchain_openai, que nos permitirão integrar facilmente os modelos DeepSeek e ChatGPT (bem como outros modelos OpenAI) no nosso projeto.

Estrutura da classe principal

Para um código mais estruturado, decidi formatá-lo como uma classe. Vamos declarar a nossa classe e descrever a lógica de inicialização:

class ChatWithAI:
    def __init__(self, provider: Literal["deepseek", "openai"] = "deepseek"):
        self.provider = provider
        self.embeddings = HuggingFaceEmbeddings(
            model_name=settings.LM_MODEL_NAME,
            model_kwargs={"device": "cuda" if torch.cuda.is_available() else "cpu"},
            encode_kwargs={"normalize_embeddings": True},
        )
        if provider == "deepseek":
            self.llm = ChatDeepSeek(
                api_key=settings.DEEPSEEK_API_KEY,
                model=settings.DEEPSEEK_MODEL_NAME,
                temperature=0.7,
            )
        elif provider == "openai":
            self.llm = ChatOpenAI(
                api_key=settings.OPENAI_API_KEY,
                model=settings.OPENAI_MODEL_NAME,
                temperature=0.7,
            )
        else:
            raise ValueError(f"Unsupported provider: {provider}")
        self.chroma_db = Chroma(
            persist_directory=settings.AMVERA_CHROMA_PATH,
            embedding_function=self.embeddings,
            collection_name=settings.AMVERA_COLLECTION_NAME,
        )

Ao criar um objeto de classe, passamos o parâmetro provider com o valor “deepseek” ou “openai” para indicar a rede neural com que iremos trabalhar no projeto.

Observe a construção:

ChatDeepSeek(
    api_key=settings.DEEPSEEK_API_KEY,
    model=settings.DEEPSEEK_MODEL_NAME,
    temperature=0.7,
)

Desta forma simples, podemos declarar a rede neural com a qual iremos interagir. Aqui, basta passar o token da API e o nome do modelo (estes são parâmetros obrigatórios). Também pode passar parâmetros adicionais para um ajuste mais fino, por exemplo, a temperatura para controlar a criatividade das respostas.

Obtenção do contexto relevante

Vamos descrever o método para obter uma resposta da base de dados:

def get_relevant_context(self, query: str, k: int = 3) -> List[Dict[str, Any]]:
    """Obtenção de contexto relevante do banco de dados."""
    try:
        results = self.chroma_db.similarity_search(query, k=k)
        return [
            {
                "text": doc.page_content,
                "metadata": doc.metadata,
            }
            for doc in results
        ]
    except Exception as e:
        logger.error(f"Erro ao obter contexto: {e}")
        return []

O método recebe como entrada o pedido do utilizador e o número de documentos que esperamos receber (por defeito, 3 documentos).

Para a pesquisa, utilizei o método similarity_search. Deixe-me lembrar-lhe que a essência deste método é que transformamos o pedido de pesquisa numa representação vetorial e, com base nesse vetor, pesquisamos na nossa base de dados.

Formatação do contexto

Vamos agora descrever o método que transformará o resultado obtido num formato conveniente para redes neuronais:

def format_context(self, context: List[Dict[str, Any]]) -> str:
    """Formatação do contexto para um prompt."""
    formatted_context = []
    for item in context:
        metadata_str = "\n".join(f"{k}: {v}" for k, v in item["metadata"].items())
        formatted_context.append(
            f"Texto: {item['text']}\nMetadata:\n{metadata_str}\n"
        )
    return "\n---\n".join(formatted_context)

This method transforms the document + metadata bundle from all received documents into one structured message, which we then pass to the neural network.

Generating a response

Now we will describe the method that will link our context with the request to the neural network:

def generate_response(self, query: str) -> Optional[str]:
    """Gere uma resposta com base num pedido e contexto."""
    try:
        context = self.get_relevant_context(query)
        if not context:
            return "Gere uma resposta com base num pedido e contexto."
        formatted_context = self.format_context(context)
        messages = [
            {
                "role": "system",
                "content": """Você é o gestor interno da empresa. Amverum Cloud (https://amverum.com/). Responde direto ao ponto, sem introduções desnecessárias.
Regras:
1.º Vá direto ao assunto, sem frases como "Com base no contexto"
2.º Use apenas factos. Se não existirem dados exatos, responda com frases gerais sobre a Nuvem Amverum, mas não dê detalhes específicos.
3.º Utilize texto simples sem formatação
4. Inclua os links apenas se estiverem em contexto
5.º Fale na primeira pessoa do plural: “Nós fornecemos”, “Nós temos”.
6.º Ao mencionar ficheiros, faça-o naturalmente, por exemplo: "Anexarei instruções onde os passos são descritos detalhadamente".
7.º Responda às saudações com gentileza e à negatividade com um pouco de humor.
8.Ao responder, pode utilizar informações gerais de fontes abertas sobre o Amverum Cloud, mas confie no contexto
9.Se o utilizador perguntar sobre preços, planos ou especificações técnicas - dê respostas específicas do contexto
10.º Para questões técnicas, ofereça soluções práticas
Personalize as respostas mencionando o nome do cliente, se estiver no contexto. Seja breve, informativo e útil.""",
 },
            {
                "role": "user",
                "content": f"Question: {query}\nContext: {formatted_context}",
            },
        ]
        response = self.llm.invoke(messages)
        if hasattr(response, "content"):
            return str(response.content)
        return str(response).strip()
    except Exception as e:
        logger.error(f"Erro ao gerar resposta:{e}")
        return "Ocorreu um erro ao gerar a resposta."

Note-se que é aqui utilizado um prompt de texto, que atua como um invólucro sobre o nosso contexto. O próprio prompt (query) pode ser qualquer coisa, e eu dei apenas um dos exemplos possíveis.

Vale a pena experimentar o conteúdo do prompt dependendo dos requisitos específicos do seu projeto. As instruções podem ser mais ou menos directivas, podem orientar o modelo para diferentes estilos de resposta ou diferentes formatos de apresentação de informação.

O método aceita uma consulta de pesquisa do utilizador como entrada. Esta consulta é primeiramente processada por uma base de dados vetorial e, com base nela, obtemos o texto estruturado dos documentos relevantes. De seguida, este texto é combinado com o prompt que preparámos e, chamando self.llm.invoke(messages), obtemos uma resposta da rede neural.

De seguida, transformamos a resposta e devolvemos ao utilizador.

Lançamento e testes

Agora só falta inicializar correctamente a classe e chamar os seus métodos:

if __name__ == "__main__":
    chat = ChatWithAI(provider="deepseek")
    print("\n=== Bate-papo com IA===\n")
    while True:
        query = input("Tu:")
        if query.lower() == "output":
            print("\nVeja o seu atraso!")
            break
        print("\nEstá a imprimir...", end="\r")
        response = chat.generate_response(query)
        print(" " * 20, end="\r")  # Limpando "Impressões de IA"..."
        print(f"AI: {response}\n")

Note que estamos aqui a executar um loop que só sairá se o utilizador digitar “exit”. Isto demonstra como podemos manter uma ligação com a base de dados de vetores do Chroma sem ter de voltar a ligar a cada pedido.

Ligamo-nos à base de dados uma vez e só nos desconectamos quando o nosso loop infinito estiver completo. Esta abordagem poupa o tempo de inicialização da ligação para cada pedido do utilizador.

Conclusão

Nesta fase, temos apenas um protótipo de um motor de busca para documentação do Amverum Cloud — é difícil chamar-lhe um serviço completo com um assistente de IA.

Se este material encontrar uma resposta em si sob a forma de visualizações, gostos e comentários, no próximo artigo direi em detalhe quais os blocos do código atual que podem ser melhorados. Com base na versão modificada, criaremos um serviço web pronto a usar, semelhante ao site chat.openai.com, onde implementaremos um chat completo e prático com o nosso assistente de IA.

Espero ter conseguido transmitir-vos como funcionam as bases de dados vetoriais, como recolher e estruturar informação para as mesmas e como organizar a geração e a pesquisa nestas bases de dados.

Devido aos recursos limitados, não consegui abordar o tema de forma profunda e abrangente. Assim, recomendo vivamente estudar a representação vetorial de dados com mais detalhe. Isto é importante pelo menos porque gigantes como o ChatGPT, Claude, DeepSeek e outras redes neuronais modernas trabalham com a representação vetorial da informação.

Além disso, recomendo que se preste atenção a uma ferramenta tão poderosa como o LangChain. O que discutimos neste artigo é apenas uma pequena parte das suas capacidades.

Isso é tudo. Vemo-nos no próximo artigo!

Alex Yacovenko

Nuvem Amverum

O Amverum Cloud é uma alternativa mais barata e fácil de implementar ao Heroku com domínios gratuitos, armazenamento persistente incluído e a possibilidade de atualizar projetos através de “git push amverum master”.

Asistente personal de IA para tus datos. Parte 1: Vector ChromaDB + DeepSeek | GPT

Amverum Cloud — Tue, 06 May 2025 05:47:21 +0000

¡Hola a todos! Hoy me gustaría abordar un tema de interés para muchos: la conexión de un modelo de lenguaje extenso como DeepSeek o ChatGPT con su base de conocimiento.

En este artículo, les explicaré detalladamente los principios de las bases de datos vectoriales y por qué pueden utilizarse para conectar su base de conocimiento con redes neuronales extensas ya preparadas.

Como ejemplo, consideremos consultar la documentación de Amverum Cloud. Amverum Cloud es una alternativa a Heroku más económica y fácil de implementar, con dominios gratuitos, almacenamiento persistente incluido y la posibilidad de actualizar proyectos mediante Git Push Amverum Master.

Actualmente estamos desarrollando activamente un agente de IA que ayudará a los usuarios a implementar proyectos en la nube, eliminando errores de código y configuración, y simplificando el trabajo con la documentación. La búsqueda en la documentación es un excelente ejemplo de cómo usar una base de datos vectorial, como Chroma.

Analicemos inmediatamente un determinado bloque limitante

Las redes neuronales preconfiguradas, como DeepSeek, ChatGPT o Claude, están diseñadas inicialmente para recibir datos de texto con un contexto específico (indicación) como entrada y, según sus parámetros y entrenamiento, procesarán la solicitud, realizando la tarea necesaria.

Es decir, dado que trabajaremos con modelos de lenguaje preconfigurados, no podremos desviarnos del proceso de preparar indicaciones con un contexto específico y enviar solicitudes a las redes neuronales preconfiguradas.

Esto nos lleva a una tarea lógica: cómo vincular nuestra base de datos con el proceso de comunicación con la red neuronal. Imaginemos miles de páginas de literatura profesional. Es improbable que convirtamos cientos de megabytes de datos en una sola indicación gigante y, al mismo tiempo, esperemos que la red neuronal acepte trabajar con ella. Como mínimo, recibiremos un error indicando que se ha excedido el contexto transmitido.

Es lógico que la solución sugiera formular la solicitud, completándola únicamente con el contexto necesario para esta solicitud específica, pero ¿cómo hacerlo? La opción más sencilla es organizar algún tipo de motor de búsqueda SEO. Enviamos, por ejemplo, "entrenamiento de redes neuronales" y se realiza una búsqueda en nuestros gigabytes de datos, pero aquí surge un problema lógico. Incluso si encontramos información, digamos 1000 veces la frase "entrenamiento de redes neuronales", ¿cómo podemos destacar este contexto? ¿Cómo decidiremos qué es importante transmitir a la red neuronal para la comunicación y qué no? Claro que se puede usar este motor de búsqueda, pero, como comprenderá, este enfoque no es del todo óptimo, lo que significa que es necesario buscar otra solución más flexible y eficaz, que analizaremos en detalle hoy.

Bases de datos vectoriales

Sé que probablemente no hayas oído hablar de estas bases de datos o solo las conozcas superficialmente. Por lo tanto, a continuación, explicaré en detalle y de forma práctica qué tipo de bases de datos son y, lo más importante, entenderemos cómo nos ayudarán en nuestra tarea.

En resumen, una base de datos vectorial es una representación de datos: texto o byte en forma numérica o, más correctamente, en forma de vectores.

Por ejemplo, la palabra "hola" en representación vectorial se vería así: -0.012, 0.124, -0.056, 0.203, …, 0.078.
En el contexto actual, no importa mucho cómo se obtienen estos números matemáticamente. Ahora solo es importante una comprensión general de la representación digital de los datos.

Beneficios prácticos

Llegados a este punto, es natural que digas: "Todo esto es pura diversión, pero ¿cuál es la utilidad práctica de todo esto y cómo se obtienen estos números tan extraños?". Vamos a averiguarlo.

Lo primero que hay que entender es que las redes neuronales no pueden leer tus textos ni ver tus fotos de forma humana. Por ejemplo, al enviar una solicitud ChatGPT, ocurre lo siguiente:

La red neuronal recibe texto.

Transforma este texto en vectores o, más correctamente, en incrustaciones.

Utiliza estas incrustaciones para generar una respuesta basada en los parámetros entrenados del modelo.

Realiza los cálculos.

Recibe una respuesta en forma de incrustación.

Transforma la incrustación en texto humano (lenguaje natural).

Te envía texto humano.

Es decir, ya en esta etapa está claro que en la comunicación con redes neuronales aparece una representación numérica de la información entrante (representación en forma de incrustaciones), y ahora veamos por qué es necesaria.

Incrustaciones y por qué son necesarias

Las incrustaciones son "huellas digitales del significado" de un texto. Analicemos esta afirmación con un ejemplo real.

Imagina que tienes una gran cantidad de documentos (libros, artículos, notas) y quieres encontrar información en ellos no por las palabras exactas, sino por su significado.

Cómo implementar esto:

La red neuronal acepta todas las palabras que has preparado y las convierte en secuencias de números (incrustaciones).

Ejemplo:
"Gato" → [0.2, -0.5, 0.7, …]
"Gatito" → 0.19, -0.48, 0.69, ….

Una base de datos vectorial, como Chroma, almacena estos números y puede:

Buscar vectores con un significado similar (incluso si las palabras son diferentes).

Responder consultas como: "Buscar algo sobre mascotas con bigote" → mostrará documentos sobre gatos y gatitos.

Analogía simple:

Búsqueda normal (como Ctrl+F): busca coincidencias exactas de palabras.

La consulta "coche" no encontrará "coche".

Búsqueda por incrustaciones: busca significados similares.

"Coche", "coche", "transporte" estarán uno al lado del otro en números, y la base de datos los conectará.

¿Por qué es esto necesario?

Los chatbots (como ChatGPT) usan incrustaciones para interpretar las consultas por significado, no por coincidencias exactas.

Puedes preguntar un documento con tus propias palabras y el sistema lo entenderá, incluso si no hay coincidencias exactas.

Resultado: Las incrustaciones son un "traductor" de texto a números, por lo que puedes buscar por significado, no por letras. Las bases de datos vectoriales son su "almacenamiento", donde la búsqueda funciona como un imán para ideas similares.

Existen redes neuronales especiales que pueden transformar tus datos de texto en una representación vectorial (numérica). Para estas tareas, no es necesario recurrir a gigantes como ChatGPT o DeepSeek, sino que bastará con redes neuronales especializadas. Además, en la parte práctica, demostraré la creación de incrustaciones y la implementación de la búsqueda inteligente utilizando el ejemplo de la red neuronal paraphrase-multilingual-MiniLM-L12-v2.

Esta red neuronal pesa solo 500 MB y se ejecuta localmente incluso en ordenadores promedio, por lo que es ideal para nuestras tareas.

Bases de datos vectoriales (continuación)

A estas alturas, ya debería comprender que, al trabajar con datos en representación numérica, se aplica la siguiente conexión:

Los datos de texto se transforman en una representación vectorial (numérica) mediante una red neuronal especial. A continuación, para que la búsqueda inteligente funcione, necesitamos:

Obtener la incrustación de la matriz de datos en representación vectorial.

Escribir una consulta de texto como "encontrar todo lo que se lleva en la mano".

Transformar la consulta en representación vectorial.

Comparar el vector de la consulta con los vectores de la incrustación grande.

Transformar el resultado encontrado de nuevo en lenguaje humano.

Proporcionar el resultado.

Ahora surge una pregunta lógica sobre el almacenamiento de estos datos numéricos. Es evidente que podemos tomar el texto y luego transformarlo en vectores utilizando una paráfrasis multilingüe MiniLM-L12-v2, aún poco clara, pero ¿qué ocurre con el almacenamiento?

Este puede almacenarse en una base de datos especializada, como Chroma, con la que trabajaremos hoy, o en RAM.

El principio del almacenamiento en RAM:

Preparar datos de texto.

Transformarlos en incrustaciones.

Enviar una solicitud de incrustación.

Una vez finalizada la ejecución del script, se borra la memoria y, al reiniciar, es necesario generar la incrustación repetidamente.

El principio del almacenamiento en una base de datos especializada:

Preparar datos de texto.

Transformarlos en incrustaciones.

Guardar la incrustación en el formato de la base de datos.

Al trabajar, conéctese a la base de datos y utilice la incrustación ya creada.

Las bases de datos vectoriales pueden ser locales (soluciones autoalojadas) o en la nube.

La base de datos local más sencilla es ChromaDB, con la que trabajaremos hoy. La elegí por su simplicidad y excelente integración con el lenguaje de programación Python.

Otras bases de datos vectoriales locales: Qdrant, Weaviate, Milvus, entre otras.

Aquí hay algunos ejemplos de bases de datos en la nube: Pinecone, RedisVL, Qdrant Cloud, entre otras.

¿Qué vamos a hacer hoy?

Habrá más teoría más adelante, pero seguro que están esperando la parte práctica. Así que, además de absorber la información teórica, veamos qué haremos en la práctica.

A continuación, les mostraré, con un ejemplo práctico, cómo transformar datos de texto en una base de datos vectorial de ChromaDB. Primero, crearemos una base de datos con un ejemplo sencillo y trabajaremos con búsquedas en ella, y luego trabajaremos con un ejemplo más complejo, sobre cuya base crearemos nuestro futuro asistente inteligente.

Después de comprender completamente cómo funcionan la generación de incrustaciones y la búsqueda inteligente, conectaremos las grandes redes neuronales DeepSeek y GPT al contexto general, implementando la siguiente lógica:

El usuario envía una solicitud específica.

Mediante la búsqueda inteligente, extraemos contenido adicional de nuestra base de datos vectorial.

Reenviaremos la solicitud del usuario junto con nuestros resultados de búsqueda inteligente a DeepSeek/GPT y recibiremos una respuesta de la red neuronal basada en nuestro contexto personal.

¡Será interesante!

Base de datos vectorial simple

Ahora, para consolidar el bloque teórico general descrito anteriormente, prepararemos una base de datos ChromaDB simple basada en la descripción de los productos de una tienda online ficticia (para el ejemplo más simple, este es un ejemplo visual).

Imaginemos que tenemos un array de productos de este tipo:

SHOP_DATA = [
    {
        "text": 'Portátil Lenovo IdeaPad 5: 16 Gb RAM, SSD 512 Gb, LED 15.6", precio 550 $.',
        "metadata": {
            "id": "1",
            "type": "product",
            "category": "laptops",
            "price": 55000,
            "stock": 3,
        },
    },
    {
        "text": 'Smartphone Xiaomi Redmi Note 12: 128 Gb RAM, SSD 512 Gb, LED 15.6", precio 1550 $.',
        "metadata": {
            "id": "2",
            "type": "product",
            "category": "phones",
            "price": 18000,
            "stock": 10,
        },
    },
    # ...
]

Como puede ver, la información se presenta como una lista de diccionarios, cada uno con dos claves principales: texto y metadatos.

Documentos y metadatos en bases de datos vectoriales

Este es un punto importante que conviene analizar con más detalle. Debe comprender que, en la base de datos ChromaDB, nuestra búsqueda inteligente funcionará de forma predeterminada solo con el valor de la clave de texto (el nombre puede ser cualquiera).

En el contexto de las bases de datos vectoriales, se distinguen conceptos como documentos (en nuestro caso, el texto que describe un producto específico) y metadatos.

Restricciones de longitud de los documentos

Al preparar los datos de texto para su carga en una base de datos vectorial, es fundamental tener en cuenta las restricciones de longitud del texto establecidas por el modelo que los convertirá en una representación vectorial. Estas restricciones dependen de la red neuronal específica y suelen oscilar entre 256 y 512 tokens. Por lo tanto, si, por ejemplo, planea agregar un libro completo de 200 páginas a la base de datos, primero debe descomponerse en fragmentos más pequeños, cada uno con un máximo de 256 a 512 tokens. Esto garantizará que el modelo procese correctamente el texto y preserve la integridad semántica en el espacio vectorial.

El modelo paraphrase-multilingual-MiniLM-L12-v2 con el que trabajaremos hoy tiene un límite de 512 tokens por documento. Esto equivale aproximadamente a 250-350 palabras o 1500-2200 caracteres. Sin embargo, dado que utilizaremos nuestros datos con las potentes redes neuronales Deepseek y ChatGPT, reduciremos el tamaño del documento para evitar la sobrecarga de contexto.

El rol de los metadatos

Los metadatos son nuestro asistente, lo que hace que la búsqueda sea más precisa y local. Aquí tiene un ejemplo sencillo:

Imaginemos que su tienda en línea vende computadoras portátiles, teléfonos inteligentes y otros equipos. Puede limitar la búsqueda inteligente por metadatos de categoría. Independientemente de si utiliza metadatos para la búsqueda explícitamente, es importante comprender que los documentos encontrados (resultados de la búsqueda) siempre contendrán estas metaetiquetas, lo que le permitirá realizar un procesamiento adicional.

Ejemplo con libros:

Imagine que decide cargar toda la serie de libros de Harry Potter en una base de datos vectorial. Divide el texto en fragmentos de 512 tokens y especifica las siguientes metaetiquetas: título del libro, número de página, número de capítulo y otra información.

A continuación, envía una consulta como "Cuando Harry Potter aprendió sobre los Horrocruxes". Su búsqueda inteligente devuelve un número determinado de resultados: documentos con metaetiquetas que le permiten comprender de qué libro proviene y en qué página se encuentra la información.

Otro ejemplo de trabajo con metaetiquetas es limitar la búsqueda a un libro, una serie de libros o un autor específico al realizar una consulta.

Recuerde que la búsqueda inteligente no está relacionada con los metadatos de ninguna manera, a menos que especifique explícitamente la lógica para procesarlos en la consulta.

Uso práctico de los metadatos
En el contexto de los metadatos para nuestros productos:

{
    "id": "1",
    "type": "product",
    "category": "laptops",
    "price": 55000,
    "stock": 3,
}

Pasamos el ID del producto, el tipo, la categoría, el precio y el stock. Por ejemplo, se podría implementar una búsqueda inteligente de productos en el sitio web y solo los productos con el ID encontrado se mostrarían al usuario en la página de resultados.

Por lo tanto, una base de datos vectorial es una herramienta excelente no solo para la integración con redes neuronales, sino también para resolver problemas prácticos, como un motor de búsqueda inteligente en un sitio web o en la API.

Creación de una base de datos vectorial en Python

Ahora escribamos código. El código será sencillo, ya que no necesitamos dividir documentos (descripciones de productos) en fragmentos ni aplicar lógica adicional para analizar y procesar productos.

Preparación del proyecto

Lo primero que hacemos es crear un nuevo proyecto de Python con un entorno virtual dedicado. Añadimos un archivo requirements.txt al proyecto con el siguiente contenido:

langchain-huggingface==0.1.2
torch==2.6.0
loguru==0.7.3
chromadb==0.6.3
sentence-transformers==3.4.1
langchain-chroma==0.2.2
pydantic-settings==2.8.1
langchain-text-splitters==0.3.7
langchain-deepseek==0.1.3
langchain-openai==0.3.11

¿Qué es Langchain?
Quizás hayas notado que he encontrado la palabra Langchain en dependencias hasta 5 veces, lo cual no es casualidad.

Langchain es una potente herramienta de Python que permite integrar una gran cantidad de redes neuronales y otras herramientas útiles en tu proyecto de forma inmediata. Este framework merece ser mencionado en más de un artículo, por lo que te recomiendo encarecidamente que leas la documentación del proyecto (más de 105 000 estrellas en GitHub deberían indicar que el proyecto merece atención).

En el proyecto de hoy, utilizaremos las siguientes herramientas de este framework:

langchain-text-splitters==0.3.7

Una herramienta para dividir textos extensos en fragmentos más pequeños (trozos). Contiene diversas estrategias de segmentación: por símbolos, oraciones, tokens y bloques semánticos. Es necesaria para procesar documentos extensos cuando superan la ventana de contexto de los modelos de lenguaje (256-512 tokens en nuestro caso).

langchain-deepseek==0.1.3

Integración con los modelos de IA de DeepSeek. Permite utilizar los potentes modelos de lenguaje de DeepSeek para el análisis y la generación de texto (hoy la demostración se basará en el modelo deepseek-chat).

langchain-openai==0.3.11

Adaptador para trabajar con la API de OpenAI. Proporciona acceso a la familia de modelos de lenguaje GPT y sus funciones. Incluye compatibilidad con modelos de chat, modelos de incrustación y funciones para trabajar con imágenes mediante DALL-E (consideremos un ejemplo con el modelo gpt-3.5-turbo).

langchain-chroma==0.2.2

Conector a la base de datos vectorial ChromaDB. Permite almacenar y recuperar eficientemente representaciones vectoriales de textos para implementar funciones de búsqueda semántica y crear bases de conocimiento con capacidad de búsqueda por similitud.

langchain-huggingface==0.1.2

La integración con el ecosistema Hugging Face brinda acceso a miles de modelos abiertos desarrollados por la comunidad. Estos incluyen modelos de lenguaje, modelos para incrustaciones, clasificación y otras tareas de PLN. Esta herramienta es especialmente útil cuando se necesita ejecutar modelos localmente o utilizar modelos especializados. En nuestro caso, podremos utilizar la red neuronal local paraphrase-multilingual-MiniLM-L12-v2.

Otros paquetes

Torch: Una biblioteca de aprendizaje automático que impulsa muchos modelos de lenguaje

Loguru: Un práctico registrador para seguir el progreso de nuestro código

ChromaDB: La base de datos vectorial que usaremos

Sentence-Transformers: Una biblioteca para convertir texto en vectores (incrustaciones)

Instalando dependencias

Ahora, instalemos:

pip install -r requirements.txt

La instalación puede tardar bastante debido al peso total de los paquetes que se instalan (en mi caso, el proceso de instalación tardó unos 20 minutos).

Creación de una base de datos

Ahora, creemos un archivo llamado create_chromadb.py (puedes darle cualquier nombre).

Realizar las importaciones:

import time
from langchain_huggingface import HuggingFaceEmbeddings
import torch
from loguru import logger
from langchain_chroma import Chroma

Definamos las variables principales:

CHROMA_PATH = "./shop_chroma_db"
COLLECTION_NAME = "shop_data"

Las colecciones en bases de datos vectoriales se pueden comparar con las tablas en bases de datos clásicas. Si no se especifica un nombre de colección, ChromaDB proporcionará un valor predeterminado.

Ahora, agreguemos una matriz de productos (cuantos más, mejor):

SHOP_DATA = [
    {
        "text": 'Lenovo IdeaPad 5: 16 RAM, SSD 512, 15.6", prise 55000 $.',
        "metadata": {
            "id": "1",
            "type": "product",
            "category": "laptops",
            "price": 55000,
            "stock": 3,
        },
    },
    # add other products
]

Y finalmente, escribamos una función para crear la base de datos:

def generate_chroma_db():
    try:
        start_time = time.time()

        logger.info("Cargando el modelo de incrustación...")
        embeddings = HuggingFaceEmbeddings(
            model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
            model_kwargs={"device": "cuda" if torch.cuda.is_available() else "cpu"},
            encode_kwargs={"normalize_embeddings": True},
        )
        logger.info(f"Modelo subido para {time.time() - start_time:.2f} segundo")

        logger.info("Creación de una base de datos de croma...")
        chroma_db = Chroma.from_texts(
            texts=[item["text"] for item in SHOP_DATA],
            embedding=embeddings,
            ids=[str(item["metadata"]["id"]) for item in SHOP_DATA],
            metadatas=[item["metadata"] for item in SHOP_DATA],
            persist_directory=CHROMA_PATH,
            collection_name=COLLECTION_NAME,
        )
        logger.info(f"Chroma DB fue creado para {time.time() - start_time:.2f} segundo")

        return chroma_db
    except Exception as e:
        logger.error(f"Error: {e}")
        raise

En este código:

Creamos un modelo para convertir texto a vectores (incrustaciones).

Inicializamos la base de datos de Chroma con nuestros productos.

Transferimos textos, IDs, metadatos y la configuración del espacio de búsqueda.

Registramos la creación correcta o los errores de la base de datos.

Es importante destacar que utilizamos el modelo multilingüe paraphrase-multilingual-MiniLM-L12-v2, compatible con el español. Esto nos permitirá crear correctamente representaciones vectoriales para nuestros productos en ruso. El modelo se cargará automáticamente desde el repositorio de Hugging Face en la primera ejecución. No será necesario cargarlo en ejecuciones posteriores si el modelo ya está instalado en su equipo.

La peculiaridad del modelo elegido es su equilibrio en tamaño y calidad: ofrece buenos resultados de vectorización de texto incluso en la CPU, pero con una tarjeta gráfica compatible con CUDA funciona mucho más rápido. Además, gracias al parámetro normalize_embeddings=True, todas las incrustaciones creadas se normalizarán, lo que mejorará la precisión de la búsqueda de productos similares.

Ejecute el código:

if __name__ == "__main__":
   generate_chroma_db()

¡Listo! Ahora tenemos la base para crear una base de datos vectorial con una descripción de los documentos. Tras el lanzamiento, en la raíz del proyecto debería aparecer una carpeta llamada "shop_chroma_db", donde se ubicarán los archivos de la base de datos. En el futuro, podremos usar esta base de datos para la búsqueda semántica de productos según las solicitudes del usuario: la búsqueda encontrará no solo coincidencias exactas, sino también productos semánticamente similares.

Creación de un motor de búsqueda de bases de datos

Ahora que la base de datos está lista, podemos describir un sistema de búsqueda simple dentro de ella. El principio será el siguiente:.

Conectarse a una base de datos existente, recibiendo incrustaciones

Recibir una consulta de búsqueda del usuario

Transformar la consulta de búsqueda en una representación vectorial

Comparar el vector de la consulta recibida con los datos de la base de datos

Devolver al usuario la cantidad de documentos que solicitó
En cuanto a la emisión de documentos al usuario, existe una función: siempre indicamos la cantidad de documentos que queremos recibir. Por ejemplo, pueden ser 5. Si no se vincula estrictamente a los metadatos mediante filtros, siempre recibirá como respuesta de la base de datos exactamente la cantidad de documentos que especificó.

Es decir, incluso si el resultado no coincide en absoluto con su solicitud, recibirá sus 5 documentos como respuesta. Es importante tener esto en cuenta al desarrollar una interfaz de usuario.

La respuesta en sí consta de: un documento, los metadatos vinculados a él y un índice de clasificación (cuanto más bajo sea el índice, mayor será la similitud en significado del documento que recibirá). La emisión se basa en el principio: cuanto más bajo sea el índice, mayor será la posición en la emisión.

Volvamos al código.

Importaciones:

from langchain_huggingface import HuggingFaceEmbeddings
import torch
from loguru import logger
from langchain_chroma import Chroma

Variables:

CHROMA_PATH = "./shop_chroma_db"
COLLECTION_NAME = "shop_data"

Ahora describamos la función para conectarse a una base de datos existente:

def connect_to_chroma():
    """Conectarse a Chroma."""
    try:
        logger.info("Descarga de incrustación...")
        embeddings = HuggingFaceEmbeddings(
            model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
            model_kwargs={"device": "cuda" if torch.cuda.is_available() else "cpu"},
            encode_kwargs={"normalize_embeddings": True},
        )

        chroma_db = Chroma(
            persist_directory=CHROMA_PATH,
            embedding_function=embeddings,
            collection_name=COLLECTION_NAME,
        )

        logger.success("Conectarse a Chroma")
        return chroma_db
    except Exception as e:
        logger.error(f"Error al conectar Chroma: {e}")
        raise

Esta función devolverá un objeto de base de datos listo para usar, dentro del cual se realizará una búsqueda vectorial.

A continuación, describiremos una función que permitirá buscar tanto por documento (páginas de documentación) como por metadatos:

def search_products(query: str, metadata_filter: dict = None, k: int = 4):
    """
    Búsqueda de páginas
    """
    try:
        chroma_db = connect_to_chroma()
        results = chroma_db.similarity_search_with_score(
            query, k=k, filter=metadata_filter
        )

        logger.info(f"Found {len(results)} query result: {query}")
        formatted_results = []
        for doc, score in results:
            formatted_results.append(
                {
                    "text": doc.page_content,
                    "metadata": doc.metadata,
                    "similarity_score": score,
                }
            )
        return formatted_results
    except Exception as e:
        logger.error(f"Error: {e}")
        raise

Y ahora llamémoslo:

for i in search_products(query="how deploy app in Amverum, with git push?"):
    print(i)

El parámetro metadata_filter nos permite especificar las condiciones de filtrado de metadatos. El ejemplo de búsqueda de productos de una tienda online es más adecuado que nuestra documentación. Por ejemplo, si queremos encontrar solo portátiles con un precio inferior a 600 $, podemos usar un filtro como este:

filter_condition = {
    "category": "laptops",
    "price": {"$lt": 600}
}

Quiero destacar que este ejemplo es educativo. En un sistema de producción, se suele crear una clase para gestionar una sesión con una base de datos, de modo que se pueda conectar una vez y mantener la conexión. En la implementación actual, cada llamada a la función reiniciará la conexión y cargará el modelo de incrustación, lo cual consume muchos recursos. En el próximo artículo, si veo que este tema les interesa, les mostraré cómo implementar dicha clase y cómo crear una aplicación web completa basada en una base de datos vectorial.

Por cierto, el código fuente del proyecto de hoy, así como la clase ya escrita para gestionar la conexión a la base de datos vectorial, se pueden encontrar en mi canal de Telegram "Easy Way to Python". El código fuente completo del proyecto lleva ahí aproximadamente una semana.

Realicen la búsqueda y vean los resultados.

Presta atención al valor de similarity_score. Cuanto menor sea este valor, más nos acercamos a la consulta en cuanto a significado.

¿No está mal, verdad? Aunque no especificamos explícitamente un modelo de aspiradora específico ni sus características en la consulta, la búsqueda vectorial logró comprender la semántica de la consulta y devolver los resultados más relevantes de nuestra base de datos. Esto demuestra el poder de las bases de datos vectoriales combinadas con modelos de lenguaje modernos.

¡A practicar!

Ahora que tienes una comprensión básica de los principios del trabajo con bases de datos vectoriales, podemos empezar a crear una lógica más compleja.

Hablaremos del servicio Amverum Cloud, que ya mencioné en mis artículos anteriores sobre Habr. En resumen, es una plataforma para la implementación sencilla (lanzamiento remoto) de proyectos en prácticamente cualquier lenguaje de programación.
Una de las grandes ventajas de Amverum Cloud es la posibilidad de implementar rápidamente un proyecto en diversas tecnologías mediante Git Push (o arrastrando archivos en la interfaz), así como obtener un nombre de dominio gratuito. Además, el servicio proporciona proxy integrado gratuito para las API de OpenAI, Antropic, Cloude y Grok, lo cual facilita la interacción con LLM. Gracias a estas capacidades técnicas, el servicio cuenta con documentación extensa y detallada para todo tipo de ocasiones: desde guías básicas, como la migración desde Heroku, hasta temas más específicos, como el lanzamiento de proyectos en FastAPI o Django, o la resolución de problemas de facturación.

Esta documentación (con aproximadamente 70 documentos diferentes) es perfecta para una demostración práctica: podemos tomar un conjunto de textos, transformarlo en una base de datos vectorial y mostrar cómo integrarla eficazmente con modelos de redes neuronales para obtener respuestas significativas y útiles.

Datos de entrada

Ya tenía en mis manos unos 70 documentos en Markdown de Amverum en este formato:

Compatibilidad con sondas de Kubernetes

¿Cómo configurar?

El formulario requiere completar la configuración en formato YAML, nativo del formato utilizado por k8s (ver aquí).
El archivo de configuración se sustituye en tu implementación, que se carga junto con él en el clúster.
Esto significa que si has cargado una configuración incorrecta o en un formato incorrecto, tu proyecto se bloqueará durante la compilación. …

Estos documentos me fueron proporcionados como un proyecto que contiene archivos MD y otros archivos técnicos. Pero lo más probable es que, en tu caso, no exista tal regalo.

Por lo tanto, es importante recordar que para recopilar información para la base de datos, puedes usar cualquier herramienta que te convenga: analizar sitios, cargar datos desde tus propias bases de datos SQL, leer documentos en Excel, Word, PDF, etc. Lo principal es que, al final, la información se convierta a un formato compatible con bases de datos vectoriales y con una estructura clara.

Lo primero que implementé en el formato de una tarea con preparación fue transformar todos estos archivos con una estructura anidada a un formato JSON de este tipo:

{
   "text": "some text",
   "metadata": {
       "section_count": 2,
       "section_1": "backups",
       "section_2": "backups /data"
   }
}

Es decir, se trata de información de texto completo en la clave de texto sin signos de puntuación, caracteres especiales ni mayúsculas, ni información técnica incluida en los metadatos, como el número de secciones, nombres, etc.

Es importante entender que los metadatos son nuestra ayuda, pero nada impediría cargar solo textos. Sin embargo, con los metadatos obtenemos oportunidades adicionales para filtrar y analizar los resultados de búsqueda.

Describamos la lógica para crear una carpeta con archivos JSON:

import json
import os
import re
import string
import sys
from typing import Any, Dict, List
from config import settings
from loguru import logger

Aquí utilicé las bibliotecas integradas de Python, con la excepción de loguru, para facilitar el registro de resultados. La extracción se basará en expresiones regulares.

A continuación, se generarán las constantes:

HEADER_PATTERN = re.compile(r"^(#+)\s(.+)")
PUNCTUATION_PATTERN = re.compile(f"[{re.escape(string.punctuation)}]")
WHITESPACE_PATTERN = re.compile(r"\s+")

Describamos la lógica para normalizar textos:

def normalize_text(text: str) -> str:
  """Normalización de texto: eliminación de puntuación y caracteres especiales."""
  if not isinstance(text, str):
  raise ValueError("El texto de entrada debe ser una cadena")

  # Eliminación de la puntuación
  text = PUNCTUATION_PATTERN.sub(" ", text)
  # Eliminar saltos de línea y espacios adicionales
  text = WHITESPACE_PATTERN.sub(" ", text)
  # Convirtiendo a minúsculas
  return text.lower().strip()

Escribamos la lógica para analizar un documento en formato md:

def parse_markdown(md_path: str) -> Dict[str, Any]:
    """Análisis de datos"""
    if not os.path.exists(md_path):
        raise FileNotFoundError(f"Archivo {md_path} no encontrado")

    try:
        with open(md_path, "r", encoding="utf-8") as file:
            content = file.read()
    except Exception as e:
        logger.error(f"Error de lectura {md_path}: {e}")
        raise

    sections: List[str] = []
    section_titles: List[str] = []
    current_section: str | None = None
    current_content: List[str] = []

    for line in content.splitlines():
        section_match = HEADER_PATTERN.match(line)

        if section_match:
            if current_section:
                sections.append("\n".join(current_content).strip())
                section_titles.append(current_section)
                current_content = []
            current_section = section_match.group(2)
            current_content.append(current_section)
        else:
            current_content.append(line)

    if current_section:
        sections.append("\n".join(current_content).strip())
        section_titles.append(current_section)

    # Text normalization
    normalized_sections = [normalize_text(section) for section in sections]
    full_text = " ".join(normalized_sections)

    # Metadata structure
    metadata = {
        "file_name": os.path.basename(md_path),
        "section_count": len(section_titles),
    }

    # Header add
    for i, title in enumerate(section_titles):
        metadata[f"section_{i+1}"] = title

    return {"text": full_text, "metadata": metadata}

Y describiremos la función para analizar todos los documentos:

def process_all_markdown(input_folder: str, output_folder: str) -> None:
    """procesamiento de rebajas"""
    if not os.path.exists(input_folder):
        raise FileNotFoundError(f"{input_folder} extraviado")

    try:
        os.makedirs(output_folder, exist_ok=True)
    except Exception as e:
        logger.error(f"Error de creación del directorio de salida: {e}")
        raise

    for root, _, files in os.walk(input_folder):
        for file_name in files:
            if file_name.endswith(".md"):
                try:
                    md_path = os.path.join(root, file_name)
                    output_path = os.path.join(
                        output_folder, file_name.replace(".md", ".json")
                    )
                    parsed_data = parse_markdown(md_path)

                    with open(output_path, "w", encoding="utf-8") as file:
                        json.dump(parsed_data, file, ensure_ascii=False, indent=4)
                    logger.info(f"Resultado en ahorro {output_path}")
                except Exception as e:
                    logger.error(f"Error de procesamiento de archivo {file_name}: {e}")

La función toma como entrada la carpeta donde se encuentran todos los documentos y la carpeta donde se escribirán los archivos JSON recibidos.

La lógica de detección de archivos md funciona buscando directorios con documentos en todas las carpetas anidadas. Esto es especialmente útil si la documentación tiene una estructura compleja de archivos y directorios.

Ahora llamamos:

if __name__ == "__main__":
    try:
        process_all_markdown(
            input_folder="ruta a la carpeta con documentos",
            output_folder="Ruta a la carpeta para guardar JSON",
        )
    except Exception as e:
        logger.error(f"Error crítico: {e}")
        sys.exit(1)

Vamos a lanzarnos.

Se recibieron 63 documentos (un ejemplo de todos los archivos está en el código fuente del proyecto)

¡Los archivos están listos! Ahora solo necesitamos leer todos los archivos y crear una base de datos vectorial basada en ellos.

Tenga en cuenta que nuestro analizador extrae no solo el texto de Markdown, sino también la estructura del documento. En los metadatos, guardamos el nombre del archivo y los encabezados de todas las secciones. Esto nos permite no solo buscar por contenido, sino también indicar con precisión de qué sección de la documentación se extrajo un fragmento en particular, lo cual es muy útil para generar respuestas significativas del asistente.

Crear una base de datos
El proceso de creación de una base de datos para Amverum no difiere mucho del enfoque descrito anteriormente, con la diferencia de que tomaremos los datos de archivos JSON existentes.

Antes de continuar, para un código más estructurado, crearemos un archivo .env en la raíz del proyecto y le asignaremos las siguientes variables:

DEEPSEEK_API_KEY=sk-1234
OPENAI_API_KEY=sk-proj-1234

Ambas claves API deben estar aquí. Si no hay claves, no es necesario crear el archivo.

A continuación, cree el archivo config.py y complételo como se indica a continuación:

import os
from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Config(BaseSettings):
    DEEPSEEK_API_KEY: SecretStr
    BASE_DIR: str = os.path.abspath(os.path.join(os.path.dirname(__file__)))
    DOCS_AMVERA_PATH: str = os.path.join(BASE_DIR, "amverum_data", "docs_amvera")
    PARSED_JSON_PATH: str = os.path.join(BASE_DIR, "amverum_data", "parsed_json")
    AMVERA_CHROMA_PATH: str = os.path.join(BASE_DIR, "amverum_data", "chroma_db")
    AMVERA_COLLECTION_NAME: str = "amvera_docs"
    MAX_CHUNK_SIZE: int = 512
    CHUNK_OVERLAP: int = 50
    LM_MODEL_NAME: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
    DEEPSEEK_MODEL_NAME: str = "deepseek-chat"
    OPENAI_MODEL_NAME: str = "gpt-3.5-turbo"
    OPENAI_API_KEY: SecretStr
    model_config = SettingsConfigDict(env_file=f"{BASE_DIR}/.env")

settings = Config()  # type: ignore

En esta clase, recopilamos todas las variables necesarias para el funcionamiento del proyecto. Como puede ver, aquí se indican las rutas a las carpetas importantes (su estructura puede variar), los nombres de las colecciones, el tamaño máximo de fragmentos y los nombres de los modelos de red neuronal.

Preste especial atención a la declaración de las claves API:

OPENAI_API_KEY: SecretStr
DEEPSEEK_API_KEY: SecretStr

Es importante describirlas no como cadenas regulares, sino mediante SecretStr. Este es un requisito de la biblioteca LangChain para el manejo seguro de datos confidenciales.

Ahora podemos importar la variable de configuración en cualquier parte del código y acceder a la configuración que necesitamos mediante un punto.

Creación de una base de datos
Volvamos a la creación de una base de datos. Comencemos con las importaciones:

import json
import os
import sys
from typing import Any, Dict, List, Optional
import torch
from langchain_chroma import Chroma
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from loguru import logger
from config import settings

Aquí, entre otras cosas, se ve la importación de RecursiveCharacterTextSplitter. Este componente nos permitirá dividir documentos grandes en fragmentos semánticos (fragmentos de texto relacionados entre sí por significado).

Escribamos una función para cargar todos los archivos JSON existentes en el formato de lista de diccionario:

def load_json_files(directory: str) -> List[Dict[str, Any]]:
    """Descarga de JSON"""
    documents = []

    try:
        if not os.path.exists(directory):
            logger.error(f"El directorio {directorio} no existe")
            return documents

        for filename in os.listdir(directory):
            if filename.endswith(".json"):
                file_path = os.path.join(directory, filename)
                try:
                    with open(file_path, "r", encoding="utf-8") as file:
                        data = json.load(file)
                        documents.append(
                            {"text": data["text"], "metadata": data["metadata"]}
                        )
                        logger.info(f"Archivo cargado: {filename}")
                except Exception as e:
                    logger.error(f"Error al leer el archivo{filename}: {e}")

        logger.success(f"Se cargaron {len(documentos)} archivos JSON")
        return documents
    except Exception as e:
        logger.error(f"Error al cargar archivos JSON: {e}")
        return documents

Tras ejecutar esta función, recibiremos todos los datos de los archivos JSON como diccionarios almacenados en RAM. La información de estos diccionarios puede utilizarse para crear una base de datos vectorial.

A continuación, describiremos la lógica para dividir documentos grandes en fragmentos significativos:

def split_text_into_chunks(text: str, metadata: Dict[str, Any]) -> List[Any]:
    """Dividir el texto en fragmentos conservando los metadatos."""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=settings.MAX_CHUNK_SIZE,
        chunk_overlap=settings.CHUNK_OVERLAP,
        length_function=len,
        is_separator_regex=False,
    )

    chunks = text_splitter.create_documents(texts=[text], metadatas=[metadata])
    return chunks

Gracias a esta función, seguimos teniendo una lista de diccionarios, pero ahora los textos extensos se dividen en fragmentos semánticos. Cada fragmento contiene los metadatos necesarios. Esto es importante para mantener el contexto general incluso al dividir un texto extenso en varias partes.

Finalmente, describiremos la lógica para crear una base de datos vectorial:

def generate_chroma_db() -> Optional[Chroma]:
    """Inicializando ChromaDB con datos de archivos JSON."""
    try:
        # Crea un directorio para almacenar la base de datos si no existe
        os.makedirs(settings.AMVERA_CHROMA_PATH, exist_ok=True)

        # Cargando archivos JSON
        documents = load_json_files(settings.PARSED_JSON_PATH)

        if not documents:
            logger.warning("No hay documentos para agregar a la base de datos.")
            return None

        # Inicializar el modelo de incrustación
        embeddings = HuggingFaceEmbeddings(
            model_name=settings.LM_MODEL_NAME,
            model_kwargs={"device": "cuda" if torch.cuda.is_available() else "cpu"},
            encode_kwargs={"normalize_embeddings": True},
        )

        # Preparación de datos para Chroma
        all_chunks = []
        for i, doc in enumerate(documents):
            chunks = split_text_into_chunks(doc["text"], doc["metadata"])
            all_chunks.extend(chunks)
            logger.info(
                f"Document {i+1}/{len(documents)} is divided into {len(chunks)} chunks"
            )

        # Crear un almacenamiento vectorial
        texts = [chunk.page_content for chunk in all_chunks]
        metadatas = [chunk.metadata for chunk in all_chunks]
        ids = [f"doc_{i}" for i in range(len(all_chunks))]

        chroma_db = Chroma.from_texts(
            texts=texts,
            embedding=embeddings,
            ids=ids,
            metadatas=metadatas,
            persist_directory=settings.AMVERA_CHROMA_PATH,
            collection_name=settings.AMVERA_COLLECTION_NAME,
            collection_metadata={
                "hnsw:space": "cosine",
            },
        )

        logger.success(
            f"Se inicializó Chroma, se agregaron {len(all_chunks)} fragmentos de {len(documents)} documentos"
        )
        return chroma_db
    except Exception as e:
        logger.error(f"Error de inicialización de Chroma: {e}")
        raise

La lógica de creación de una base de datos no difiere mucho de los ejemplos que vimos anteriormente. La principal diferencia radica en la división de textos extensos en fragmentos significativos (trozos).

En mi modesto ordenador, el proceso de creación de una base de datos tardó menos de 10 minutos. En dispositivos con GPU, este proceso solo tardará unos minutos gracias a la computación paralela.

Optimización del rendimiento

Cabe destacar que, al trabajar con grandes volúmenes de datos, se puede optimizar aún más el proceso:

Utilizar el procesamiento por lotes al crear incrustaciones (procesamiento de documentos en grupos).

Configurar los parámetros MAX_CHUNK_SIZE y CHUNK_OVERLAP según la estructura de los datos.

Elegir un modelo de incrustación adecuado, buscando un equilibrio entre calidad y velocidad.

ChromaDB ofrece un buen rendimiento para la mayoría de las tareas RAG, pero para volúmenes de datos muy grandes (millones de documentos), conviene considerar otras soluciones, como FAISS o Pinecone.

Integración de un motor de búsqueda con redes neuronales

Anteriormente, vimos cómo describir un motor de búsqueda sin utilizar grandes redes neuronales. Para consolidar este material, te sugiero que crees tú mismo un motor de búsqueda basado en tu base de datos.

Pasaré a la parte más interesante: explicaré cómo combinar la salida de una base de datos vectorial con redes neuronales para crear un agente de IA inteligente. Para ello, utilizaremos datos de la documentación de Amverum Cloud, que ya he transformado en una base de datos vectorial.

Configuración inicial del proyecto

Comencemos creando un archivo llamado chat_with_ai.py.

Realicemos las importaciones necesarias:

from typing import Any, Dict, List, Literal, Optional
import torch
from config import settings
from langchain_chroma import Chroma
from langchain_deepseek import ChatDeepSeek
from langchain_openai import ChatOpenAI
from langchain_huggingface import HuggingFaceEmbeddings
from loguru import logger

Aquí incluimos las nuevas herramientas langchain_deepseek y langchain_openai, que nos permitirán integrar fácilmente los modelos DeepSeek y ChatGPT (así como otros modelos de OpenAI) en nuestro proyecto.

Estructura de la clase principal

Para un código más estructurado, decidí formatearlo como una clase. Declaremos nuestra clase y describamos la lógica de inicialización:

class ChatWithAI:
    def __init__(self, provider: Literal["deepseek", "openai"] = "deepseek"):
        self.provider = provider
        self.embeddings = HuggingFaceEmbeddings(
            model_name=settings.LM_MODEL_NAME,
            model_kwargs={"device": "cuda" if torch.cuda.is_available() else "cpu"},
            encode_kwargs={"normalize_embeddings": True},
        )

        if provider == "deepseek":
            self.llm = ChatDeepSeek(
                api_key=settings.DEEPSEEK_API_KEY,
                model=settings.DEEPSEEK_MODEL_NAME,
                temperature=0.7,
            )
        elif provider == "openai":
            self.llm = ChatOpenAI(
                api_key=settings.OPENAI_API_KEY,
                model=settings.OPENAI_MODEL_NAME,
                temperature=0.7,
            )
        else:
            raise ValueError(f"Proveedor no compatible: {provider}")

        self.chroma_db = Chroma(
            persist_directory=settings.AMVERA_CHROMA_PATH,
            embedding_function=self.embeddings,
            collection_name=settings.AMVERA_COLLECTION_NAME,
        )

Al crear un objeto de clase, pasamos el parámetro proveedor con el valor "deepseek" o "openai" para indicar la red neuronal con la que trabajaremos en el proyecto.

Observe la construcción:

ChatDeepSeek(
    api_key=settings.DEEPSEEK_API_KEY,
    model=settings.DEEPSEEK_MODEL_NAME,
    temperature=0.7,
)

De esta forma sencilla, podemos declarar la red neuronal con la que interactuaremos. Aquí, basta con pasar el token de la API y el nombre del modelo (estos son parámetros obligatorios). También se pueden pasar parámetros adicionales para un ajuste más preciso, por ejemplo, la temperatura para controlar la creatividad de las respuestas.

Obtención del contexto relevante
Describamos el método para obtener una respuesta de la base de datos:

def get_relevant_context(self, query: str, k: int = 3) -> List[Dict[str, Any]]:
    """Obtener contexto relevante de la base de datos."""
    try:
        results = self.chroma_db.similarity_search(query, k=k)
        return [
            {
                "text": doc.page_content,
                "metadata": doc.metadata,
            }
            for doc in results
        ]
    except Exception as e:
        logger.error(f"Error al obtener el contexto: {e}")
        return []

El método toma como entrada la solicitud del usuario y el número de documentos que esperamos recibir (por defecto, 3).
Para la búsqueda, utilicé el método similarity_search. Cabe recordar que la esencia de este método es transformar la solicitud de búsqueda en una representación vectorial y, con base en este vector, buscar en nuestra base de datos.

Formatear el contexto
A continuación, describiremos el método que transformará el resultado obtenido en un formato adecuado para redes neuronales:

def format_context(self, context: List[Dict[str, Any]]) -> str:
    """Dar formato al contexto de un mensaje."""
    formatted_context = []
    for item in context:
        metadata_str = "\n".join(f"{k}: {v}" for k, v in item["metadata"].items())
        formatted_context.append(
            f"Текст: {item['text']}\nMetadata:\n{metadata_str}\n"
        )
    return "\n---\n".join(formatted_context)

Este método transforma el paquete de documentos y metadatos de todos los documentos recibidos en un mensaje estructurado, que luego pasamos a la red neuronal.

Generación de una respuesta
Ahora describiremos el método que vinculará nuestro contexto con la solicitud a la red neuronal:

def generate_response(self, query: str) -> Optional[str]:
    """Generar una respuesta basada en una solicitud y contexto."""
    try:
        context = self.get_relevant_context(query)
        if not context:
            return "Lo siento, no pude encontrar el contexto relevante para la respuesta."

        formatted_context = self.format_context(context)

        messages = [
            {
"role": "system",
"content": """Eres el administrador interno de la empresa. Amverum Cloud (https://amverum.com/). Respondes concisamente, sin preámbulos.
Reglas:
1. Vaya directo al grano, sin frases como "Basado en el contexto".
2. Use solo datos. Si no hay datos exactos, responda con frases generales sobre Amverum Cloud, pero no entre en detalles.
3. Use texto plano sin formato.
4. Incluya enlaces solo si están en contexto.
5. Hable en primera persona del plural: "Proporcionamos", "Tenemos".
6. Al mencionar archivos, hágalo con naturalidad, por ejemplo: "Adjuntaré instrucciones donde se detallan los pasos".
7. Responda a los saludos con amabilidad, a las negativas con un poco de humor.
8. Al responder, puede usar información general de fuentes abiertas sobre Amverum Cloud, pero apóyese en el contexto.
9. Si el usuario pregunta sobre precios, planes o especificaciones técnicas, proporcione respuestas específicas del contexto.
10. Para preguntas técnicas, ofrezca soluciones prácticas.

Personalice las respuestas mencionando el nombre del cliente si está en el contexto. Sea breve, informativo y útil.""",
 },
            {
                "role": "user",
                "content": f"Pregunta: {query}\nContext: {formatted_context}",
            },
        ]
        response = self.llm.invoke(messages)
        if hasattr(response, "content"):
            return str(response.content)
        return str(response).strip()
    except Exception as e:
        logger.error(f"Error al generar respuesta: {e}")
        return "Se produjo un error al generar la respuesta."

Tenga en cuenta que aquí se utiliza un mensaje de texto, que actúa como contenedor de nuestro contexto. El mensaje en sí (consulta) puede ser cualquier cosa, y solo he dado un ejemplo posible.

Conviene experimentar con el contenido del mensaje según los requisitos específicos de su proyecto. Las instrucciones pueden ser más o menos directivas y orientar el modelo a diferentes estilos de respuesta o formatos de presentación de la información.

El método acepta una consulta de búsqueda del usuario como entrada. Esta consulta se procesa primero mediante una base de datos vectorial y, a partir de ella, obtenemos el texto estructurado de los documentos relevantes. A continuación, este texto se combina con el mensaje que hemos preparado y, al llamar a self.llm.invoke(messages), obtenemos una respuesta de la red neuronal.

A continuación, transformamos la respuesta y la devolvemos al usuario.

Lanzamiento y pruebas
Ahora solo queda inicializar correctamente la clase y llamar a sus métodos:

if __name__ == "__main__":
    chat = ChatWithAI(provider="deepseek")
    print("\n=== Chat with AI===\n")

    while True:
        query = input("Tú: ")
        if query.lower() == "output":
            print("\nVeo tu tarde!")
            break

        print("\nImpresión con IA...", end="\r")
        response = chat.generate_response(query)
        print(" " * 20, end="\r")  # Borrar "Impresiones AI""..."
        print(f"AI: {response}\n")

Tenga en cuenta que estamos ejecutando un bucle que solo saldrá si el usuario pulsa "exit". Esto demuestra cómo podemos mantener la conexión con la base de datos vectorial de Chroma sin tener que reconectarnos con cada solicitud.

Nos conectamos a la base de datos una vez y solo nos desconectamos al completar el bucle infinito. Este enfoque ahorra el tiempo de inicializar la conexión con cada solicitud del usuario.

Conclusión

En esta etapa, solo tenemos un prototipo de motor de búsqueda para la documentación de Amverum Cloud; es difícil considerarlo un servicio completo con un asistente de IA.

Si este material genera interés en forma de visualizaciones, "me gusta" y comentarios, en el próximo artículo les detallaré qué bloques del código actual se pueden mejorar. Basándonos en la versión modificada, crearemos un servicio web listo para usar similar al sitio web chat.openai.com, donde implementaremos un chat completo y práctico con nuestro asistente de IA. Espero haberles explicado cómo funcionan las bases de datos vectoriales, cómo recopilar y estructurar información para ellas, y cómo organizar la generación y la búsqueda en dichas bases de datos.

Debido a la limitación de recursos, no pude abordar el tema a fondo. Por lo tanto, recomiendo encarecidamente estudiar la representación vectorial de datos con más detalle. Esto es importante, sobre todo porque gigantes como ChatGPT, Claude, DeepSeek y otras redes neuronales modernas trabajan con la representación vectorial de la información.
Además, recomiendo prestar atención a una herramienta tan potente como LangChain. Lo que hemos analizado en este artículo es solo una pequeña parte de sus capacidades.

Eso es todo. ¡Nos vemos en el próximo artículo!

Alex Yacovenko

Amverum Cloud

Amverum Cloud es una alternativa a Heroku más económica y fácil de implementar, con dominios gratuitos, almacenamiento persistente incluido y la posibilidad de actualizar proyectos mediante "git push amverum master".