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

Выбор базы данных для Telegram-проекта сильно зависит от типа бота и ожидаемой нагрузки. Бот для уведомлений одного сервиса и мультиплеерная игра-MiniApp с 100,000 пользователей требуют принципиально разных решений. Разберём все сценарии с примерами кода.

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

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

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

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

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

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

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

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

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

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

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 postgres from 'postgres'
import { validateTelegramInitData } from './telegram-auth'

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

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 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}"
    pipe = redis_client.pipeline()
    pipe.incr(key)
    pipe.expire(key, window)
    results = await pipe.execute()
    count = results[0]
    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 бот: бэкап каждые 30 минут
dbsend source add \
  --name="telegram-bot" \
  --db=sqlite:///opt/bot/bot.db \
  --schedule="*/30 * * * *" \
  --keep=200

# PostgreSQL бот: бэкап каждый час
dbsend source add \
  --name="telegram-bot-pg" \
  --db="${DATABASE_URL}" \
  --schedule="0 * * * *" \
  --keep=72

# Проверка
dbsend status

FAQ

SQLite или PostgreSQL для начинающего разработчика Telegram-бота?

Начните с SQLite — нулевая настройка, не нужен отдельный сервер, проще локальная разработка. SQLite справится с нагрузкой до 5,000-10,000 пользователей при правильном WAL-режиме. Когда почувствуете ограничения (конкурентная запись, сложные запросы) — мигрируйте на PostgreSQL. Aiogram поддерживает оба варианта.

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

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

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

Да, для небольших объёмов. Структура: таблица messages с полями user_id, role (user/assistant), content (TEXT), created_at. При запросе к AI извлекайте последние N сообщений. При больших объёмах (>1M сообщений) лучше PostgreSQL с партиционированием таблицы по дате.

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

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

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

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