Бот с базой данных PostgreSQL: архитектура

MAATRIX

Бот на SQLite нормально живёт, пока пользователей десятки, а данных — горстка записей в одном файле. Но стоит проекту начать расти — параллельные запросы, история действий, аналитика, несколько инстансов бота — и SQLite начинает капризничать блокировками, а код, где SQL вперемешку с обработчиками команд, превращается в кашу. Ниже — рабочая схема архитектуры бота с PostgreSQL, которая не разваливается через полгода: таблицы, пул соединений, миграции и разделение слоёв.

Почему PostgreSQL, а не SQLite

SQLite — файл на диске с блокировкой на запись: один писатель одновременно, остальные ждут. Для бота с сотней сообщений в минуту это терпимо, но как только вы добавляете фоновые задачи (рассылки, крон-джобы, обработку вебхуков), блокировки начинают конфликтовать и вы ловите database is locked в самый неподходящий момент.

PostgreSQL решает это через MVCC: читатели не блокируют писателей, а параллельные транзакции над разными строками не мешают друг другу. Плюс нормальные типы данных (jsonb, timestamptz, массивы), полноценные внешние ключи, оконные функции для аналитики и возможность разнести бота и базу на разные машины, когда один сервер перестанет тянуть нагрузку.

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

Структура таблиц: пользователи, история, настройки

Начните с трёх сущностей — этого достаточно для 90% ботов, дальше расширяете по мере надобности.

CREATE TABLE users (
    id            BIGINT PRIMARY KEY,        -- telegram_id / discord_id
    username      TEXT,
    full_name     TEXT,
    locale        TEXT NOT NULL DEFAULT 'ru',
    is_blocked    BOOLEAN NOT NULL DEFAULT FALSE,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE user_settings (
    user_id       BIGINT PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
    notifications BOOLEAN NOT NULL DEFAULT TRUE,
    timezone      TEXT NOT NULL DEFAULT 'Europe/Moscow',
    preferences   JSONB NOT NULL DEFAULT '{}'::jsonb
);

CREATE TABLE action_log (
    id            BIGSERIAL PRIMARY KEY,
    user_id       BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    action        TEXT NOT NULL,             -- 'command', 'callback', 'payment' и т.д.
    payload       JSONB,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX idx_action_log_user_created ON action_log (user_id, created_at DESC);
CREATE INDEX idx_action_log_action ON action_log (action);

Ключевые решения здесь: user_id — реальный ID из Telegram/Discord, а не свой автоинкремент — так проще джойнить и не нужен отдельный маппинг. Настройки вынесены в отдельную таблицу «один к одному» с users, а не размазаны по колонкам — появится десятое поле настроек, не придётся переписывать основную таблицу. preferences JSONB — клапан для гибких настроек, которые не хочется выносить в отдельные колонки ради двух пользователей, которые ими пользуются.

action_log — лог событий, а не мутируемое состояние: только INSERT, никаких UPDATE. Такая таблица растёт быстро, поэтому индекс по (user_id, created_at DESC) обязателен — иначе выборка «последние 20 действий пользователя» превратится в full scan через пару месяцев. Если лог станет по-настоящему большим (десятки миллионов строк), имеет смысл партиционировать его по месяцам через PARTITION BY RANGE (created_at), но на старте это преждевременная оптимизация.

Пул соединений: почему нельзя коннектиться на каждый запрос

Самая частая ошибка новичков — открывать новое соединение с БД на каждый запрос:

# Так делать не надо — соединение на каждый вызов
async def get_user(user_id: int):
    conn = await asyncpg.connect(DSN)
    row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
    await conn.close()
    return row

Установка TCP-соединения с PostgreSQL — это handshake, аутентификация, выделение процесса на стороне сервера (PostgreSQL по умолчанию форкает процесс на каждое соединение). При десятках запросов в секунду это заметная задержка на каждый вызов и риск упереться в max_connections (по умолчанию 100), если бот получает всплеск активности.

Правильный подход — пул соединений, который создаётся один раз при старте и живёт всё время работы бота:

import asyncpg

async def create_pool() -> asyncpg.Pool:
    return await asyncpg.create_pool(
        dsn=DATABASE_URL,
        min_size=5,
        max_size=20,
        command_timeout=10,
        max_inactive_connection_lifetime=300,
    )

# при старте бота
pool = await create_pool()

# в обработчике
async def get_user(pool: asyncpg.Pool, user_id: int):
    async with pool.acquire() as conn:
        return await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)

min_size держит горячие соединения наготове, max_size — потолок, который не даст боту исчерпать лимит соединений на сервере БД. Ориентировочно: для одного инстанса бота средней нагрузки 10-20 соединений в пуле хватает с запасом; точное число зависит от трафика и стоит подбирать по метрикам, а не угадывать. Пул передаётся в обработчики через dependency injection фреймворка (в aiogram 3 — через middleware или workflow_data, в discord.py — через атрибут бота), а не создаётся заново в каждом хендлере.

Если бот работает через несколько процессов (вебхук за несколькими воркерами uvicorn/gunicorn), у каждого процесса свой пул — не пытайтесь шарить asyncpg.Pool между процессами, он привязан к event loop.

Миграции схемы через Alembic

Ручные ALTER TABLE, которые вы применяете «руками через psql» — гарантированный способ рассинхронизировать схему между дев-окружением и продакшеном. Alembic решает это версионированными миграциями поверх SQLAlchemy Core (модели ORM для этого не обязательны).

Установка и инициализация:

pip install alembic asyncpg sqlalchemy
alembic init -t async migrations

В migrations/env.py укажите строку подключения через переменную окружения, а не хардкодом:

config.set_main_option("sqlalchemy.url", os.environ["DATABASE_URL"])

Создание миграции — вручную (для контроля над SQL) или автогенерацией по моделям:

alembic revision -m "add user_settings table"

# migrations/versions/xxxx_add_user_settings_table.py
def upgrade() -> None:
    op.create_table(
        "user_settings",
        sa.Column("user_id", sa.BigInteger, sa.ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
        sa.Column("notifications", sa.Boolean, nullable=False, server_default="true"),
        sa.Column("timezone", sa.Text, nullable=False, server_default="Europe/Moscow"),
        sa.Column("preferences", sa.dialects.postgresql.JSONB, nullable=False, server_default="{}"),
    )

def downgrade() -> None:
    op.drop_table("user_settings")

Применение на сервере — часть деплоя, а не отдельный ручной шаг:

alembic upgrade head

Держите миграции в git вместе с кодом бота и запускайте alembic upgrade head в скрипте деплоя перед перезапуском процесса. downgrade() пишите честно, даже если откатывать «не планируете» — рано или поздно понадобится откатить неудачный релиз в 3 часа ночи, и лучше, чтобы это была одна команда, а не восстановление из бэкапа.

Разделение бизнес-логики и работы с БД

Когда SQL-запросы разбросаны прямо по хендлерам команд, тестировать логику без реальной базы невозможно, а любое изменение схемы требует грепать весь проект. Стандартное решение — слой репозиториев между обработчиками и БД.

# repositories/users.py
class UserRepository:
    def __init__(self, pool: asyncpg.Pool):
        self._pool = pool

    async def get_or_create(self, user_id: int, username: str | None) -> asyncpg.Record:
        async with self._pool.acquire() as conn:
            async with conn.transaction():
                row = await conn.fetchrow(
                    "SELECT * FROM users WHERE id = $1", user_id
                )
                if row is None:
                    row = await conn.fetchrow(
                        """
                        INSERT INTO users (id, username)
                        VALUES ($1, $2)
                        RETURNING *
                        """,
                        user_id, username,
                    )
                return row

    async def log_action(self, user_id: int, action: str, payload: dict | None = None) -> None:
        async with self._pool.acquire() as conn:
            await conn.execute(
                "INSERT INTO action_log (user_id, action, payload) VALUES ($1, $2, $3)",
                user_id, action, payload,
            )

# handlers/start.py — бизнес-логика ничего не знает про SQL
async def handle_start(message: types.Message, users: UserRepository):
    user = await users.get_or_create(message.from_user.id, message.from_user.username)
    await users.log_action(user["id"], "command", {"name": "start"})
    await message.answer(f"Привет, {user['username'] or 'друг'}!")

Репозиторий отвечает только за SQL: он не знает про Telegram, про формат сообщений, про бизнес-правила. Хендлер не знает про структуру таблиц — он вызывает методы с понятными именами. Разделение окупается на втором-третьем месяце жизни проекта: UserRepository можно протестировать отдельно (например, с тестовой БД в Docker), а логику команд — с замоканным репозиторием, без реального PostgreSQL в юнит-тестах.

Для проектов покрупнее между репозиторием и хендлером добавляют сервисный слой (UserService) с бизнес-правилами («заблокировать после трёх нарушений», «начислить бонус за приглашение») — но для типичного бота двух слоёв обычно достаточно, третий добавляйте по реальной необходимости, а не заранее.

Индексы, транзакции и типичные грабли под нагрузкой

Несколько вещей, которые не видны на старте, но больно бьют при росте:

• N+1 запросы. Цикл по списку пользователей с вызовом get_user на каждого — это N запросов вместо одного WHERE id = ANY($1). При рассылке на тысячу пользователей разница ощутима.
• Транзакции вокруг связанных операций. Если создание пользователя и запись первого действия должны быть атомарны — оборачивайте их в async with conn.transaction(), а не полагайтесь на порядок выполнения двух INSERT.
• **SELECT * в горячих путях.** Для таблиц с jsonb-полями явное перечисление нужных полей снижает объём передаваемых данных и защищает от сюрпризов при изменении схемы.
• Отсутствие индекса под частый фильтр. EXPLAIN ANALYZE на медленных запросах — первое, что стоит проверить, прежде чем добавлять кэш или расширять сервер.
• Долгие транзакции. Ожидание ответа от внешнего API внутри блока conn.transaction() держит блокировки и соединение из пула — сетевые вызовы лучше делать вне транзакций.

PostgreSQL любит оперативную память под кэш страниц и shared_buffers, а для бота с активным логом действий пригодится SSD с приличным IOPS — на медленных сетевых дисках INSERT-heavy нагрузка от action_log может стать узким местом раньше, чем ожидаете. Если бот и база живут на одном сервере: сколько ресурсов нужно VPS для Telegram-ботов, а для разнесения по разным машинам — какой VPS лучше для базы данных.

Частые вопросы