База данных для Telegram бота и MiniApp: какую выбрать в 2026 году

Выбор базы данных для Telegram-проекта зависит от типа бота и характера нагрузки. Бот уведомлений для одного сервиса и MiniApp-игра с конкурентными покупками и рейтингом требуют разных решений. Ниже — четыре типовых сценария, код для aiogram и Node.js и настройка бэкапов.

Типы Telegram-проектов и их требования

Простой бот (напоминания, подписки, уведомления):

• Данные: user_id, настройки, состояние диалога
• Нагрузка: низкая (<100 RPS)
• Решение: SQLite достаточно

Чат-бот с AI (GPT-wrapper, обработка документов):

• Данные: история диалогов, файлы, токены
• Нагрузка: средняя, пики при запросах
• Решение: PostgreSQL + Redis для кэша

MiniApp — игра или SaaS (рейтинги, транзакции, социальные функции):

• Данные: пользователи, баллы, история, лидерборд
• Нагрузка: высокая, конкурентная запись
• Решение: PostgreSQL + Redis

MiniApp — e-commerce (каталог, корзина, заказы):

• Данные: продукты, пользователи, заказы, платежи
• Нагрузка: средняя, критична целостность данных
• Решение: PostgreSQL (ACID-транзакции)

SQLite: для простых ботов без сервера

SQLite — хороший старт для простого бота: нет сервера, нет настройки, файл рядом с кодом. Включите WAL-режим (PRAGMA journal_mode=WAL), чтобы чтение не блокировалось записью. Подробнее о самой СУБД — в статье что такое SQLite.

Python + aiogram 3 + aiosqlite

# pip install aiogram aiosqlite

import asyncio
import aiosqlite
from aiogram import Bot, Dispatcher, types
from aiogram.filters import CommandStart

DB_PATH = "bot.db"

async def init_db():
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY,
                user_id INTEGER UNIQUE NOT NULL,
                username TEXT,
                first_name TEXT,
                language TEXT DEFAULT 'ru',
                notifications_enabled BOOLEAN DEFAULT 1,
                created_at DATETIME DEFAULT CURRENT_TIMESTAMP
            )
        """)
        await db.execute("""
            CREATE TABLE IF NOT EXISTS reminders (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER NOT NULL,
                text TEXT NOT NULL,
                remind_at DATETIME NOT NULL,
                sent BOOLEAN DEFAULT 0,
                FOREIGN KEY (user_id) REFERENCES users(id)
            )
        """)
        await db.commit()

async def get_or_create_user(user: types.User) -> dict:
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        # Upsert — создать или обновить
        await db.execute("""
            INSERT INTO users (user_id, username, first_name)
            VALUES (?, ?, ?)
            ON CONFLICT(user_id) DO UPDATE SET
                username = excluded.username,
                first_name = excluded.first_name
        """, (user.id, user.username, user.first_name))
        await db.commit()
        
        async with db.execute(
            "SELECT * FROM users WHERE user_id = ?", (user.id,)
        ) as cursor:
            return dict(await cursor.fetchone())

# Инициализация бота
bot = Bot(token="YOUR_BOT_TOKEN")
dp = Dispatcher()

@dp.message(CommandStart())
async def start(message: types.Message):
    user = await get_or_create_user(message.from_user)
    await message.answer(
        f"Привет, {user['first_name']}! Я запомнил тебя 👋"
    )

async def main():
    await init_db()
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())

Node.js + telegraf + better-sqlite3

// npm install telegraf better-sqlite3 @types/better-sqlite3

import { Telegraf } from 'telegraf'
import Database from 'better-sqlite3'

const db = new Database('bot.db')
db.pragma('journal_mode = WAL')
db.pragma('foreign_keys = ON')

// Инициализация схемы
db.exec(`
  CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY,
    telegram_id INTEGER UNIQUE NOT NULL,
    username TEXT,
    name TEXT,
    points INTEGER DEFAULT 0,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
  );
  
  CREATE TABLE IF NOT EXISTS actions (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id INTEGER REFERENCES users(id),
    action TEXT NOT NULL,
    metadata TEXT, -- JSON
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
  );
`)

// Prepared statements для производительности
const getUser = db.prepare('SELECT * FROM users WHERE telegram_id = ?')
const createUser = db.prepare(
  'INSERT OR IGNORE INTO users (telegram_id, username, name) VALUES (?, ?, ?)'
)
const addPoints = db.prepare(
  'UPDATE users SET points = points + ? WHERE telegram_id = ?'
)

const bot = new Telegraf(process.env.BOT_TOKEN!)

bot.start((ctx) => {
  const { id, username, first_name } = ctx.from
  createUser.run(id, username ?? null, first_name)
  const user = getUser.get(id) as any
  ctx.reply(`Привет! У тебя ${user.points} баллов 🎯`)
})

bot.launch()

PostgreSQL: для продакшн-бота и MiniApp

PostgreSQL — выбор для серьёзного проекта. ACID-транзакции, конкурентная запись, оконные функции для лидербордов.

Python + aiogram 3 + asyncpg

# pip install aiogram asyncpg python-dotenv

import asyncpg
import asyncio
from aiogram import Bot, Dispatcher

DATABASE_URL = "postgresql://user:pass@host:5432/botdb"

# Connection pool — переиспользование соединений
pool: asyncpg.Pool = None

async def create_pool():
    global pool
    pool = await asyncpg.create_pool(
        DATABASE_URL,
        min_size=5,
        max_size=20,
        command_timeout=10,
    )
    
    # Создание таблиц при старте
    async with pool.acquire() as conn:
        await conn.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id BIGINT PRIMARY KEY,           -- telegram user_id
                username TEXT,
                first_name TEXT NOT NULL,
                language_code VARCHAR(10) DEFAULT 'ru',
                is_premium BOOLEAN DEFAULT false,
                balance DECIMAL(10,2) DEFAULT 0,
                referral_id BIGINT REFERENCES users(id),
                registered_at TIMESTAMPTZ DEFAULT NOW()
            );
            
            CREATE TABLE IF NOT EXISTS transactions (
                id SERIAL PRIMARY KEY,
                user_id BIGINT REFERENCES users(id),
                amount DECIMAL(10,2) NOT NULL,
                type VARCHAR(50) NOT NULL,  -- deposit, withdraw, bonus
                description TEXT,
                created_at TIMESTAMPTZ DEFAULT NOW()
            );
            
            CREATE INDEX IF NOT EXISTS idx_transactions_user_id 
                ON transactions(user_id);
        """)

async def get_or_create_user(user_id: int, username: str, first_name: str) -> dict:
    async with pool.acquire() as conn:
        # UPSERT — атомарная операция
        row = await conn.fetchrow("""
            INSERT INTO users (id, username, first_name)
            VALUES ($1, $2, $3)
            ON CONFLICT (id) DO UPDATE SET
                username = EXCLUDED.username,
                first_name = EXCLUDED.first_name
            RETURNING *
        """, user_id, username, first_name)
        return dict(row)

async def add_points_atomic(user_id: int, amount: float, description: str):
    """Атомарное добавление баллов + запись в историю"""
    async with pool.acquire() as conn:
        async with conn.transaction():
            await conn.execute(
                "UPDATE users SET balance = balance + $1 WHERE id = $2",
                amount, user_id
            )
            await conn.execute("""
                INSERT INTO transactions (user_id, amount, type, description)
                VALUES ($1, $2, 'bonus', $3)
            """, user_id, amount, description)

async def get_leaderboard(limit: int = 10) -> list:
    async with pool.acquire() as conn:
        rows = await conn.fetch("""
            SELECT 
                first_name,
                balance,
                RANK() OVER (ORDER BY balance DESC) as rank
            FROM users
            ORDER BY balance DESC
            LIMIT $1
        """, limit)
        return [dict(r) for r in rows]

Telegram MiniApp: архитектура бэкенда

MiniApp — React/Vue приложение внутри Telegram. Бэкенд обычно REST или GraphQL API.

// backend/src/routes/miniapp.ts — Node.js/Bun.js + Fastify

import Fastify from 'fastify'
import { drizzle } from 'drizzle-orm/postgres-js'
import { eq, sql } from 'drizzle-orm'
import postgres from 'postgres'
import { users, shopItems, inventory } from './schema'
import { validateTelegramInitData } from './telegram-auth'

const client = postgres(process.env.DATABASE_URL!)
const db = drizzle(client)

const app = Fastify()

// Middleware: проверка Telegram InitData
app.addHook('preHandler', async (request, reply) => {
    const initData = request.headers['x-telegram-init-data'] as string
    
    if (!initData) {
        return reply.status(401).send({ error: 'No auth' })
    }
    
    const user = validateTelegramInitData(initData, process.env.BOT_TOKEN!)
    if (!user) {
        return reply.status(401).send({ error: 'Invalid auth' })
    }
    
    // Прикрепляем пользователя к запросу
    request.telegramUser = user
})

// Endpoint: профиль пользователя
app.get('/api/profile', async (request) => {
    const { id, first_name, username } = request.telegramUser
    
    // Upsert в PostgreSQL
    const [user] = await db
        .insert(users)
        .values({ telegramId: BigInt(id), firstName: first_name, username })
        .onConflictDoUpdate({
            target: users.telegramId,
            set: { firstName: first_name, username },
        })
        .returning()
    
    return { user }
})

// Endpoint: купить предмет (транзакция)
app.post('/api/shop/buy', async (request, reply) => {
    const { itemId } = request.body as { itemId: number }
    const userId = BigInt(request.telegramUser.id)
    
    // Атомарная транзакция
    try {
        await db.transaction(async (tx) => {
            const [user] = await tx
                .select({ balance: users.balance })
                .from(users)
                .where(eq(users.telegramId, userId))
                .for('update') // FOR UPDATE — блокировка строки
            
            const [item] = await tx
                .select()
                .from(shopItems)
                .where(eq(shopItems.id, itemId))
            
            if (user.balance < item.price) {
                throw new Error('Недостаточно баллов')
            }
            
            // Списание баллов
            await tx.update(users)
                .set({ balance: sql`balance - ${item.price}` })
                .where(eq(users.telegramId, userId))
            
            // Добавление предмета в инвентарь
            await tx.insert(inventory).values({ userId, itemId })
        })
        
        return { success: true }
    } catch (e: any) {
        return reply.status(400).send({ error: e.message })
    }
})

Redis: сессии, rate limiting и кэш

Redis решает задачи, которые PostgreSQL делает медленно или неудобно:

# pip install redis[asyncio]

import json
import redis.asyncio as redis

redis_client = redis.from_url("redis://localhost:6379", decode_responses=True)

# 1. Состояние FSM (Finite State Machine) для aiogram
async def set_user_state(user_id: int, state: str, ttl: int = 3600):
    await redis_client.setex(f"state:{user_id}", ttl, state)

async def get_user_state(user_id: int) -> str | None:
    return await redis_client.get(f"state:{user_id}")

# 2. Rate limiting (защита от спама)
async def check_rate_limit(user_id: int, action: str, limit: int, window: int) -> bool:
    key = f"ratelimit:{user_id}:{action}"
    count = await redis_client.incr(key)
    if count == 1:
        # TTL ставим только на первый запрос, иначе окно будет сдвигаться бесконечно
        await redis_client.expire(key, window)
    return count <= limit  # True = разрешено

# Использование
@dp.message()
async def handle_message(message: types.Message):
    allowed = await check_rate_limit(
        message.from_user.id, 
        "message", 
        limit=5,    # 5 сообщений
        window=60   # в минуту
    )
    if not allowed:
        await message.answer("⏱ Подождите немного...")
        return
    # обработка...

# 3. Кэш лидерборда (обновляется каждые 5 минут)
async def get_cached_leaderboard():
    cached = await redis_client.get("leaderboard")
    if cached:
        return json.loads(cached)
    
    # Запрос к PostgreSQL — дорогой
    leaderboard = await get_leaderboard()
    await redis_client.setex("leaderboard", 300, json.dumps(leaderboard))
    return leaderboard

Архитектура по размеру проекта

Ориентиры, а не жёсткие пороги — реальный предел зависит от доли записей и тяжести запросов:

| Проект | БД | Redis | Хостинг | |---|---|---|---| | Простой бот (<1,000 users) | SQLite | Нет | VPS 512MB | | Средний бот (<10,000 users) | PostgreSQL | Нет (опционально) | VPS 1-2GB | | Большой бот / MiniApp | PostgreSQL | Да | VPS 4GB+ | | Highload MiniApp | PostgreSQL + реплика | Redis Cluster | Облако |

Бэкапы для Telegram-бота

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

Для SQLite не копируйте файл через cp на работающем боте: в WAL-режиме часть транзакций лежит в bot.db-wal. Используйте онлайн-бэкап:

# SQLite: консистентная копия работающей базы
OUT="/var/backups/bot-$(date +%F-%H%M).db"
sqlite3 /opt/bot/bot.db ".backup '$OUT'"
sqlite3 "$OUT" "PRAGMA quick_check;"

Для PostgreSQL — pg_dump. Он делает согласованный снимок и не блокирует обычные чтение и запись:

pg_dump --format=plain --no-owner "$DATABASE_URL" > /var/backups/telegram-bot.sql

Расписание через cron или systemd timer, выгрузку в S3 и ротацию старых копий смотрите в статьях бэкап SQLite Telegram-бота и pg_dump по cron в S3.

Если не хотите сами вести хранилище и ротацию, копии можно отправлять в dbsend. CLI сам делает консистентную копию SQLite перед загрузкой, а дамп PostgreSQL вы создаёте на своей стороне:

# SQLite: например, из cron каждые 30 минут
dbsend backup /opt/bot/bot.db -d "$DBSEND_SQLITE_ID" -e sqlite

# PostgreSQL: дамп + загрузка, например, раз в час
pg_dump --format=plain --no-owner "$DATABASE_URL" > /tmp/telegram-bot.sql
dbsend backup /tmp/telegram-bot.sql -d "$DBSEND_POSTGRES_ID" -e postgresql_dump

# Проверка ключа и история версий
dbsend doctor
dbsend log "$DBSEND_SQLITE_ID"

Если файл не изменился с прошлой загрузки, новая версия не создаётся. На платных тарифах dbsend пришлёт алерт в Telegram, если бэкап не пришёл по расписанию. Подробнее — на странице бэкапы SQLite.

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

SQLite или PostgreSQL для первого Telegram-бота?

Начните с SQLite: нулевая настройка, не нужен отдельный сервер, проще локальная разработка. В WAL-режиме SQLite спокойно обслуживает типичного бота с одним процессом. Переходите на PostgreSQL, когда появляются несколько процессов или серверов, пишущих в базу, конкурентные транзакции с деньгами или сложная аналитика. aiogram работает с обоими вариантами.

Что такое Telegram InitData и зачем его проверять?

InitData — строка с данными пользователя (id, имя, username и др.), которую Telegram передаёт в MiniApp при запуске. Данные не зашифрованы, а подписаны: подпись HMAC-SHA256 вычисляется с ключом, производным от токена бота. Проверка подписи на сервере подтверждает, что данные пришли от Telegram. Без неё любой может подставить чужой user_id.

Можно ли хранить историю диалогов с AI в SQLite?

Да. Таблица messages с полями user_id, role (user/assistant), content и created_at, индекс по (user_id, created_at). При запросе к модели берите последние N сообщений. Когда объём и число пишущих процессов растут, переносите историю в PostgreSQL.

Нужен ли Redis для aiogram FSM?

aiogram 3 поддерживает несколько хранилищ состояний: MemoryStorage (по умолчанию, теряется при перезапуске), RedisStorage и MongoStorage. Для продакшн-бота используйте постоянное хранилище, чаще всего RedisStorage, — иначе пользователи теряют контекст диалога при каждом деплое.

Какую ORM использовать для Telegram-бота на Python?

SQLAlchemy 2 в async-режиме — гибкая, поддерживает SQLite (через aiosqlite) и PostgreSQL (через asyncpg). Tortoise ORM — проще в настройке, изначально асинхронная. Для простого синхронного бота подойдёт peewee. В Node.js — Drizzle ORM: он работает и с better-sqlite3, и с postgres.js.