Уровень: middle/senior backend Стек: FastAPI 0.115, SQLite, WebSocket, slowapi, JWT-like sessions, Docker Что внутри: как у меня работает бэкенд мессенджера, почему SQLite вместо Postgres, монолит на 19к строк, самописные миграции, sliding expiration для токенов
Преамбула
Это восьмая статья из моей серии про инженерные решения в ONEMIX. До этого было про клиентскую часть мессенджера: кэш сообщений, E2E, WebRTC звонки, Electron, outbox-паттерн. Параллельно про AI-агента Лиру и мнение про вайб-кодинг.
Сегодня про серверную сторону. Backend ONEMIX — это один файл main.py на 19603 строки, 379 эндпоинтов, FastAPI + SQLite, держит мессенджер с регистрацией через SMS, звонками через LiveKit, E2E через Double Ratchet, push-нотификациями на iOS и Android. Этот файл я пишу больше года. За это время он эволюционировал из прототипа на 800 строк в production монолит.
В статье разберу почему стек именно такой, какие решения оказались правильными, какие я бы поменял если бы начинал сейчас, и где у этого подхода границы применимости.
Сразу важная оговорка. У меня не было требования держать 100к одновременных пользователей или 10к RPS. Это бэкенд под мобильное приложение с трафиком который для соло-разработчика разумно поддерживать одному. Если у вас задачи другого масштаба, мой опыт может не подойти.
Стек одним взглядом
# requirements.txtfastapi==0.115.0uvicorn[standard]==0.30.6psycopg2-binary==2.9.9 # для будущей миграции, сейчас не используетсяsqlite3 # из stdlib, основная БДpython-dotenv==1.0.1pydantic==2.8.2python-multipart==0.0.9httpx==0.27.0slowapi==0.1.9pillow==10.4.0pillow-heif==0.18.0aiofiles==24.1.0websockets==13.0PyJWT==2.9.0cryptography==43.0.1
15 зависимостей. Никаких ORM (SQLAlchemy нет), никакого Celery, никакого Redis. Pydantic для валидации входящих JSON, slowapi для rate limiting, всё остальное стандартное.
Docker-compose состоит из двух сервисов: backend контейнер и Postgres. Тут уже первый интересный момент. Postgres подключён, но в коде используется SQLite. Объясню зачем дальше.
SQLite в production
Если посмотреть main.py, увидите вот это:
import sqlite3BASE_DIR = os.path.abspath(os.path.dirname(__file__))DATABASE_PATH = os.path.join(BASE_DIR, "messenger.db")def get_db_connection(): conn = sqlite3.connect(DATABASE_PATH, check_same_thread=False) conn.row_factory = sqlite3.Row return conn
Это production, не локальная разработка, не тесты, не прототип. SQLite держит весь мессенджер: пользователи, чаты, сообщения, реакции, звонки, сессии. На отдельном файле messenger.db.
Почему SQLite вместо Postgres для production — стандартный вопрос. Прежде чем отвечать — Postgres у меня готов. Docker-compose поднимает его параллельно, psycopg2-binary в requirements.txt, есть документ POSTGRES_MIGRATION.md с планом переезда. Я могу мигрировать когда захочу. Просто пока не нужно.
Почему SQLite справляется
В мессенджере 90% записей это INSERT новых сообщений и SELECT по chat_id. Никаких сложных JOIN’ов между шардами, никакой аналитики в реальном времени, никаких отчётов по огромным таблицам. SQLite на современных SSD держит десятки тысяч INSERT в секунду на одном файле. Этого хватает с большим запасом.
WAL-режим (Write-Ahead Logging) включается одной строчкой и позволяет читать БД параллельно с записью. Никаких lock contention которого все боятся. На моих данных типичная операция занимает 0.1-2мс.
Backup — это cp messenger.db messenger.db.bak. Без mysqldump, без pg_dump, без отдельной инфраструктуры. Файл скопировался — backup готов.
Где SQLite не справится
Я не делаю вид что SQLite это серебряная пуля. Где она точно проиграет:
Многопроцессовая запись. Если у вас 5 uvicorn-воркеров пишут одновременно — будет lock contention. Я держу один воркер с asyncio, и этого хватает.
Сетевой доступ к БД. SQLite это файл, нельзя поделить между разными машинами. Если масштабироваться горизонтально — нужен сетевой DB.
Сложная аналитика. Если хочется crosstable JOIN’ы по миллионам строк — Postgres лучше, SQLite это OLTP-only.
Postgres-специфичные фичи. JSON-операторы, GIS, full-text search через pg_trgm — это всё богаче чем SQLite. Если нужно — Postgres.
Когда я мигрирую
Триггер для миграции это не «пользователей стало больше», а «появилась задача которую SQLite не закрывает». Пока её нет.
Если завтра нужно будет масштабировать горизонтально или появится сложная аналитика — мигрирую за неделю. План записан, инфраструктура готова.
Монолит на 19603 строки
Второй неочевидный выбор — один файл main.py на 19603 строки с 379 эндпоинтами. Не папка с модулями, не разделение по доменам, не APIRouter’ы.
Это сознательно. Объясню почему.
Когда разрабатываешь и поддерживаешь бэкенд в одиночку, главный ресурс это скорость навигации в коде. Когда нужно поменять эндпоинт /chats/{id}/messages, нужно знать где он лежит. В монолите ответ один: в main.py. В микросервисах или мульти-модульной структуре — нужно помнить какая папка, какой router, какой файл.
С Cmd+P в IDE я открываю main.py за секунду. С Cmd+Shift+O (Go to Symbol) нахожу любой эндпоинт по имени функции, их 379. Эта навигация работает плохо в мультимодульных проектах. Имена коллизионируют между модулями, нужно ходить по папкам.
Что я делаю чтобы 19к строк не превратились в хаос:
Логические разделители блочными комментариями. Группы эндпоинтов отделены такими блоками:
# ============ AUTH ============# ============ CHATS ============# ============ MESSAGES ============
В IDE с поддержкой code folding это работает почти как папки.
Stand-alone эндпоинты. Каждая функция-эндпоинт независима, никаких общих хелперов на полстатьи. Читаю эндпоинт — вся его логика в нём же, не нужно прыгать по 5 файлам.
Pydantic-модели в одном блоке в начале файла. Все request/response схемы лежат вместе, легко найти.
Постоянный grep. Когда не помню где что, делаю grep "chats/.*messages" main.py. Найдёт за секунду.
Когда монолит сломается. Если бы я нанимал второго backend-разработчика, монолит был бы плохим решением. Два человека в одном файле дают merge-конфликты на каждом коммите. Я бы тогда разбил по доменам с APIRouter и пакетами. Сейчас один разработчик и один файл, это оптимально.
Самописные миграции без alembic
Стандартный совет для FastAPI + БД, поставить alembic. Я не поставил, миграции делаю сам:
def add_column_if_missing(table_name, column_name, column_type, index=False): cursor.execute(f"PRAGMA table_info({table_name})") columns = [row[1] for row in cursor.fetchall()] if column_name not in columns: print(f"[DB] Migrating {table_name}: adding {column_name} column") try: cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_type}") if index: cursor.execute(f"CREATE UNIQUE INDEX IF NOT EXISTS idx_{table_name}_{column_name} ON {table_name}({column_name})") conn.commit() except Exception as e: print(f"[DB] Migration error ({table_name}.{column_name}): {e}")# Применениеadd_column_if_missing('users', 'phone', 'TEXT', index=True)add_column_if_missing('users', 'username', 'TEXT', index=True)add_column_if_missing('users', 'bio', 'TEXT')add_column_if_missing('chats', 'is_private', 'INTEGER DEFAULT 0')# ... ещё 50 таких строк
Это простой self-healing подход. При запуске сервера init_db() проходит по всем нужным колонкам и добавляет недостающие. SQLite поддерживает ALTER TABLE ADD COLUMN без блокировок.
Почему не alembic. Alembic это отдельный инструмент со своей CLI, своей нумерацией миграций, своими отдельными файлами. Чтобы добавить колонку нужно создать migration file, прописать upgrade/downgrade, запустить alembic upgrade head, не забыть закоммитить migration file. Это работает в команде где нужна формальность. Для одиночного разработчика это overhead.
С моим подходом проще. Добавил строку add_column_if_missing, рестартанул сервер, готово. Дев-цикл в 3 раза короче.
Где это сломается. Если нужны сложные миграции с переименованием колонок, конвертацией данных, разделением таблиц, нужен инструмент посерьёзнее. У меня таких миграций пока не было. Когда будет — поставлю alembic ретроспективно или мигрирую на Postgres + готовый инструмент.
И я не утверждаю что это правильный подход для всех. Это работает для моего профиля задач. Если у вас в команде 5 человек делают миграции, alembic обязателен, иначе хаос гарантирован.
Sessions вместо JWT со sliding expiration
Третий неочевидный выбор. Auth не через JWT, а через таблицу sessions в БД.
def get_current_user(request, authorization, db): if not authorization: raise HTTPException(401, "Not authenticated") token = authorization.replace("Bearer ", "") cursor = db.cursor() cursor.execute(""" SELECT s.user_id, s.expires_at, u.id, u.name, u.phone, ... FROM sessions s JOIN users u ON s.user_id = u.id WHERE s.token = ? """, (token,)) row = cursor.fetchone() if not row: raise HTTPException(401, "Invalid token") expires_at = datetime.fromisoformat(row["expires_at"].replace('Z', '+00:00')) if datetime.now(timezone.utc) > expires_at: raise HTTPException(401, "Token expired") # Sliding expiration — продлеваем срок жизни активного токена days_left = (expires_at - now_dt).total_seconds() / 86400 if days_left < 7: new_expires = (now_dt + timedelta(days=30)).isoformat() cursor.execute("UPDATE sessions SET expires_at = ? WHERE token = ?", (new_expires, token)) return {"id": row["id"], "name": row["name"], ...}
JWT не использую. Вместо этого таблица sessions, где token это просто случайная строка, которая хранится в БД и сверяется на каждом запросе.
Что я получаю по сравнению с JWT:
Можно отозвать токен. JWT отозвать нельзя без отдельного blacklist’а. Sessions просто DELETE FROM sessions WHERE token = ? и всё, пользователь сразу разлогинен.
Можно увидеть активные сессии. SELECT по user_id и юзер видит все свои устройства, может отключить ненужные.
Sliding expiration. Это интересная штука. Стандартная JWT-схема: токен живёт 1 час, потом refresh-token меняет на новый. У меня по-другому. Токен живёт 30 дней изначально. Каждый раз когда юзер делает запрос, проверяется сколько до истечения. Если меньше 7 дней, токен продлевается ещё на 30. Активный юзер никогда не разлогинивается. Неактивный через 30 дней без активности.
Цена этого подхода — JOIN с users на каждом запросе. У меня это занимает 0.5-2мс на SQLite, не критично.
Когда стоит JWT. Если у вас несколько микросервисов которые должны проверять токен без обращения к centralized auth-БД, JWT даёт stateless проверку. У меня один монолит, проблемы stateless нет.
WebSocket connection manager
Real-time через WebSocket напрямую в FastAPI, без отдельного Redis Pub/Sub. Все соединения держатся в памяти одного процесса.
class ConnectionManager: def __init__(self): self.active_connections: Dict[str, Set[WebSocket]] = {} self.online_users: Set[str] = set() # Лок ТОЛЬКО для мутаций словаря — не удерживается во время IO self._conn_lock = asyncio.Lock() async def send_personal_message(self, message: dict, user_id: str): # Снимок соединений без удержания лока на время IO connections = set(self.active_connections.get(user_id, set())) if not connections: return dead: Set[WebSocket] = set() for connection in connections: try: await connection.send_json(message) except Exception: dead.add(connection) if dead: async with self._conn_lock: bucket = self.active_connections.get(user_id) if bucket: bucket -= dead async def broadcast_to_users(self, message: dict, user_ids: List[str]): # Параллельная рассылка через asyncio.gather await asyncio.gather( *(self.send_personal_message(message, uid) for uid in user_ids), return_exceptions=True )
Ключевая штука. Лок только на мутацию словаря, не на IO. Если держать async with self._conn_lock пока идёт await connection.send_json(message), все broadcasts сериализуются. Один зависший клиент блокирует рассылку всем остальным.
Поэтому делаю снимок соединений в локальную переменную под локом, отпускаю лок, и параллельно шлю всем без удержания лока. Если соединение мёртвое, добавляю в dead set, и в конце под локом удаляю.
broadcast_to_users через asyncio.gather с return_exceptions=True. Это параллельная рассылка всем участникам группы. На 50 участников группового чата это 1-3мс вместо 50мс последовательной рассылки. Один упавший WebSocket не роняет остальные, exceptions ловятся и игнорируются.
Когда это сломается. Если запустить 2+ uvicorn воркеров, каждый имеет свой ConnectionManager в памяти. Юзер подключён к воркеру A, broadcast идёт через воркер B, юзер не получит сообщение. Тут нужен Redis Pub/Sub между воркерами или Postgres LISTEN/NOTIFY. У меня один воркер, поэтому нет проблемы.
pts: серверная сторона sync state
Для синхронизации состояния клиента и сервера использую pts (Points of Tracking State), как в MTProto Telegram. Я уже разбирал клиентскую сторону в статье про кэш (ссылка). Здесь покажу серверную.
def _get_next_pts(cursor, chat_id: str) -> int: """Атомарно инкрементирует pts для чата и возвращает новое значение.""" cursor.execute(""" INSERT INTO chat_pts (chat_id, pts) VALUES (?, 1) ON CONFLICT(chat_id) DO UPDATE SET pts = pts + 1 """, (chat_id,)) cursor.execute("SELECT pts FROM chat_pts WHERE chat_id = ?", (chat_id,)) row = cursor.fetchone() return row["pts"] if row else 1def pts_increment(cursor, chat_id: str, event: str, data: dict) -> int: """Единая точка входа: увеличивает pts, пишет лог.""" pts = _get_next_pts(cursor, chat_id) _log_pts_event(cursor, chat_id, pts, event, data) return pts
pts это монотонно возрастающий счётчик событий чата. При каждом изменении (новое сообщение, удаление, реакция, редактирование) pts растёт на 1. Клиент хранит свой pts и при реконнекте запрашивает getDifference(pts=N). Сервер возвращает все события после этого pts.
Это надёжнее timestamp. Timestamp может совпасть у двух событий, может перескочить из-за смены времени на сервере, может быть неточным из-за округления. pts гарантирует порядок и отсутствие пропусков.
INSERT ... ON CONFLICT DO UPDATE это атомарная операция в SQLite (и в Postgres, синтаксис тот же). Без транзакций, без блокировок, без race condition между двумя одновременными INSERT.
Если интересно как это работает на клиенте, там сложнее. Нужно сохранять pts в локальный кэш и при холодном старте восстанавливать состояние без полной синхронизации. Это разобрано в первой статье серии про кэш.
Что бы я сделал по-другому
Что бы я сделал по-другому, если бы начинал сейчас:
Сразу разделил бы на 2-3 файла. Не на полноценные модули, а просто auth.py, chats.py, admin.py плюс main.py с включением. 19к строк в одном файле всё-таки много. 8-10к на файл было бы оптимально.
Не делал бы интеграции в монолит. У меня в одном файле и мессенджер, и интеграции с Botemy, и Lyra-bridge, и admin-панель. По-хорошему это отдельные сервисы, потому что они отдельно деплоятся, отдельно ломаются, отдельно нагружаются. Сейчас если падает один эндпоинт админки, это может задеть мессенджер.
Поставил бы Postgres сразу. Не для производительности, а чтобы потом не делать миграцию данных. SQLite работает, но при миграции 30 ГБ accumulated data это отдельный квест. Лучше начать с Postgres даже если можно было обойтись SQLite.
Поставил бы Sentry или похожий error tracking с первого дня. Сейчас у меня логи в stdout, и я хожу по ним глазами. На production-нагрузке это не работает. Sentry или Glitchtip это вечер настройки и потом годы спокойной жизни.
Когда этот стек НЕ подойдёт
Реалистично, где мой подход применим, а где нет.
Применим: соло-разработчик или команда до 3 человек, мобильное приложение или веб с разумной нагрузкой (тысячи RPS максимум), монолитный домен (один продукт), нет требований 99.99% uptime.
Не применим: команда от 5 человек, требования к horizontal scaling, нагрузка десятки тысяч RPS, несколько отдельных продуктов с шарингом БД, требования соответствия (GDPR, HIPAA) с extensive audit logging.
Если ваш профиль из первой категории, мой стек разумный выбор. Если из второй, нужны микросервисы, Kubernetes, Postgres с репликацией, и команда DevOps. Это другая лига.
Заключение
Главный тезис. Простой стек который вы понимаете обычно работает лучше сложного стека который рекомендуют все. SQLite вместо Postgres не для всех, но для соло-разработчика часто оптимально. Монолит вместо микросервисов не для команды, но для одного человека идеально. Самописные миграции вместо alembic, это компромисс между скоростью и формальностью.
Я часто вижу как соло-разработчики мучительно настраивают Kubernetes, Redis-кластер, RabbitMQ для проекта который держит 50 пользователей. Это карго-культ, не инженерия. Правильный инструмент тот, который закрывает реальную задачу с минимальной сложностью, а не тот который называют «стандартом индустрии».
В Telegram-канале пишу мелкие наблюдения по горячему. Недавно были посты про race condition в outbox, attention degradation в длинных промптах. Иногда привожу куски кода прямо со скриншотами IDE. Если интересно: https://t.me/SkillTrackr.
Это восьмая статья из серии. В предыдущих был мобильный клиент ONEMIX: трёхуровневый кэш, Double Ratchet E2E, WebRTC звонки, vanilla Electron, outbox-паттерн. А также мнение про вайб-кодинг и AI-агент Лира. Эта про серверную сторону.
Если интересны конкретные куски — как именно устроен SMS-auth flow, как pts работает на клиенте, как поднять Postgres-миграцию без даунтайма — пишите, разберу в комментариях или следующей статье.
ссылка на оригинал статьи https://habr.com/ru/articles/1034814/