Database

Database (База данных)

Система использует SQLite в качестве основной СУБД через async SQLAlchemy + aiosqlite. Три независимых файла БД:

База данных	Путь	Назначение
Основная	`data/secretary.db`	Пользователи, сессии, чаты, боты, виджеты, CRM, канбан и т.д.
Telegram Sales	`data/sales.db` (single) / `data/sales_{bot_id}.db` (multi)	Воронка продаж Telegram-бота (пользователи, события, discovery, платежи)
WhatsApp Sales	`data/wa_sales.db` (single) / `data/wa_sales_{instance_id}.db` (multi)	Воронка продаж WhatsApp-бота (пользователи, события, discovery)

Миграция legacy-путей: До v1.1 Telegram sales DB по умолчанию создавалась в корне проекта (sales.db). Теперь все БД хранятся в data/. При старте бота старый файл автоматически мигрирует в data/sales.db вместе с WAL/SHM сайдкарами.

PRAGMA-инициализация

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

PRAGMA	Значение	Назначение
`journal_mode`	`WAL`	Конкурентное чтение во время записи (вместо полной блокировки)
`foreign_keys`	`ON`	Включение проверки внешних ключей и `ON DELETE CASCADE`
`busy_timeout`	`5000`	5 секунд ожидания при блокировке (вместо мгновенной ошибки)
`synchronous`	`NORMAL`	Оптимальный баланс скорости и надёжности для WAL-режима

Основная БД (SQLAlchemy)

PRAGMA устанавливаются через event listener на engine.sync_engine:

# db/database.py
@event.listens_for(engine.sync_engine, "connect")
def _set_sqlite_pragmas(dbapi_conn, connection_record):
    cursor = dbapi_conn.cursor()
    cursor.execute("PRAGMA journal_mode=WAL")
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.execute("PRAGMA busy_timeout=5000")
    cursor.execute("PRAGMA synchronous=NORMAL")
    cursor.close()

С StaticPool (единственное постоянное соединение) listener срабатывает ровно 1 раз.

Sales БД (aiosqlite)

PRAGMA устанавливаются напрямую в SalesDatabase.init():

# telegram_bot/sales/database.py, whatsapp_bot/sales/database.py
await self._db.execute("PRAGMA journal_mode=WAL")
await self._db.execute("PRAGMA foreign_keys=ON")
await self._db.execute("PRAGMA busy_timeout=5000")
await self._db.execute("PRAGMA synchronous=NORMAL")

WAL-режим

Write-Ahead Logging (WAL) — режим журналирования, который позволяет читателям не блокироваться при записи. Особенности:

При WAL появляются дополнительные файлы: secretary.db-wal и secretary.db-shm
При корректном завершении все три БД выполняют PRAGMA wal_checkpoint(TRUNCATE):
- Основная: close_db() в db/database.py
- Sales (Telegram/WhatsApp): SalesDatabase.close() в соответствующих sales/database.py
Бэкапы безопасны: BackupService автоматически вызывает WAL checkpoint перед копированием каждого .db-файла и включает -wal/-shm сайдкары как safety net (см. Backup)

Health Check

Эндпоинт GET /health возвращает информацию о БД, включая PRAGMA-конфигурацию:

{
  "database": {
    "status": "ok",
    "path": "/app/data/secretary.db",
    "size_mb": 2.45,
    "journal_mode": "wal",
    "foreign_keys_enabled": true
  }
}

Если journal_mode не "wal" или foreign_keys_enabled не true — это признак проблемы с инициализацией.

Unit of Work (транзакции)

Основная БД использует паттерн Unit of Work: репозитории только делают flush() (отправляют SQL в БД без коммита), а коммит выполняет вызывающий код — менеджер (db/integration.py), роутер или сервис.

┌──────────────────────────────────────────┐
│ Caller (manager / router / service)      │
│  async with AsyncSessionLocal() as s:    │
│      repo = SomeRepo(s)                  │
│      await repo.create(entity)  # flush  │
│      await repo.update(other)   # flush  │
│      await session.commit()     # ← ✅   │
└──────────────────────────────────────────┘

Зачем?

Атомарность: несколько операций коммитятся как одна транзакция — либо все, либо ни одна
Консистентность: данные не попадают в БД частично при ошибке между операциями
Контроль: вызывающий код решает, когда фиксировать изменения

Правила

Слой	`flush()`	`commit()`	`rollback()`
Репозитории (`db/repositories/`)	Да	Нет	Нет
BaseRepository (`create`, `update`, `delete`)	Да (автоматически)	Нет	Нет
Менеджеры (`db/integration.py`)	Нет	Да (после всех операций)	Нет (обрабатывается session context)
Роутеры (`app/routers/`)	Нет	Да (write-блоки с `AsyncSessionLocal`)	Нет
`get_async_session()` (FastAPI DI)	Нет	Да (автоматически при успехе)	Да (при исключении)
`get_session_context()`	Нет	Да (автоматически при успехе)	Да (при исключении)

Пример

# db/repositories/user.py — только flush
async def create(self, entity):
    self.session.add(entity)
    await self.session.flush()  # SQL INSERT, но не зафиксирован
    return entity

# db/integration.py — коммит в менеджере
async def create_user(self, ...):
    async with AsyncSessionLocal() as session:
        repo = UserRepository(session)
        user = await repo.create(User(...))
        await session.commit()  # ← фиксация транзакции
        return user.to_dict()

Антипаттерны

# ❌ НЕЛЬЗЯ: коммит в репозитории
async def create(self, entity):
    self.session.add(entity)
    await self.session.commit()  # нарушает Unit of Work

# ❌ НЕЛЬЗЯ: забыть коммит в роутере
async with AsyncSessionLocal() as session:
    repo = SomeRepo(session)
    await repo.create(entity)
    # нет commit → данные потеряются!

Миграции

Две системы миграций:

Система	Использование
Alembic (предпочтительно)	`alembic upgrade head` — новые миграции
Legacy скрипты	`python scripts/migrate_*.py` — старые миграции

Новые таблицы создаются автоматически через Base.metadata.create_all при старте. Изменения схемы существующих таблиц требуют миграционных скриптов.

# Применить все миграции
alembic upgrade head

# Создать новую миграцию из изменений моделей
alembic revision --autogenerate -m "description"

# Посмотреть историю миграций
alembic history

Диагностика

Проверка PRAGMA

# Внутри Docker
docker exec ai-secretary python -c "
import asyncio
from db.database import AsyncSessionLocal, init_db
from sqlalchemy import text

async def check():
    await init_db()
    async with AsyncSessionLocal() as s:
        for p in ['journal_mode', 'foreign_keys', 'busy_timeout', 'synchronous']:
            r = (await s.execute(text(f'PRAGMA {p}'))).scalar()
            print(f'{p} = {r}')

asyncio.run(check())
"

Ожидаемый вывод:

journal_mode = wal
foreign_keys = 1
busy_timeout = 5000
synchronous = 1

Проверка целостности

sqlite3 data/secretary.db "PRAGMA integrity_check;"
# Ответ: ok

Проверка FK

sqlite3 data/secretary.db "PRAGMA foreign_key_check;"
# Пустой вывод = нет нарушений FK

Тесты

28 тестов в tests/unit/:

pytest tests/unit/ -v -o asyncio_mode=auto

Unit of Work тесты (`test_unit_of_work.py`)

Тест	Что проверяет
`test_repo_does_not_commit`	Repo flush без commit → rollback теряет данные
`test_repo_flush_visible_in_session`	Flush делает данные видимыми в пределах сессии
`test_commit_persists_data`	Commit сохраняет данные после закрытия сессии
`test_multi_op_atomicity`	Несколько операций коммитятся/откатываются атомарно

PRAGMA тесты (`test_sqlite_pragmas.py`)

Тест	Что проверяет
`test_main_engine_pragmas`	Все 4 PRAGMA значения после init
`test_pragmas_persist_across_sessions`	PRAGMA сохраняются между сессиями (StaticPool)
`test_foreign_keys_enforced`	INSERT с невалидным FK вызывает IntegrityError
`test_wal_checkpoint_on_close`	WAL checkpoint очищает WAL-файл
`test_event_listener_fires_once_with_static_pool`	Listener вызывается ровно 1 раз
`test_telegram_sales_db_pragmas`	PRAGMA в Telegram sales DB
`test_whatsapp_sales_db_pragmas`	PRAGMA в WhatsApp sales DB
`test_sales_db_fk_enforced`	FK enforcement в sales DB

Бэкап тесты (`test_backup_safety.py`)

Тест	Что проверяет
`test_checkpoint_wal_clears_wal_file`	WAL checkpoint обнуляет WAL-файл
`test_checkpoint_wal_nonexistent_db`	Checkpoint на несуществующей БД не падает
`test_backup_includes_sales_dbs`	ZIP содержит main DB + все sales DBs
`test_backup_includes_wal_sidecars`	ZIP содержит непустые WAL/SHM сайдкары
`test_backup_skips_empty_sidecars`	Пустые сайдкары не включаются
`test_manifest_version_1_1`	Manifest version = 1.1
`test_restore_cleans_stale_wal`	Restore удаляет stale WAL/SHM
`test_restore_handles_sales_dbs`	Restore распаковывает sales DBs
`test_restore_creates_bak_files`	Restore создаёт .bak копии
`test_get_system_info_includes_sales_dbs`	system_info содержит sales DBs

Sales DB тесты (`test_sales_db_close.py`)

Тест	Что проверяет
`test_telegram_sales_close_checkpoints`	close() делает WAL checkpoint
`test_whatsapp_sales_close_checkpoints`	close() делает WAL checkpoint
`test_telegram_sales_default_path`	Default = `data/sales.db`
`test_whatsapp_sales_default_path`	Default = `data/wa_sales.db`
`test_telegram_legacy_migration`	`sales.db` мигрирует в `data/sales.db`
`test_telegram_close_idempotent`	Двойной close() не падает

← Backup | Wiki-RAG →

Database

Database (База данных)

PRAGMA-инициализация

Основная БД (SQLAlchemy)

Sales БД (aiosqlite)

WAL-режим

Health Check

Unit of Work (транзакции)

Зачем?

Правила

Пример

Антипаттерны

Миграции

Диагностика

Проверка PRAGMA

Проверка целостности

Проверка FK

Тесты

Unit of Work тесты (test_unit_of_work.py)

PRAGMA тесты (test_sqlite_pragmas.py)

Бэкап тесты (test_backup_safety.py)

Sales DB тесты (test_sales_db_close.py)

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Навигация

Главная

Мониторинг

ИИ и модели

Каналы

Бизнес

Система

Интеграции

Безопасность

Unit of Work тесты (`test_unit_of_work.py`)

PRAGMA тесты (`test_sqlite_pragmas.py`)

Бэкап тесты (`test_backup_safety.py`)

Sales DB тесты (`test_sales_db_close.py`)