awg-dash/backend/awg_api/store/db.py

"""Открытие и жизненный цикл SQLite-хранилища состояния (issue #20, ADR-006).

Путь по умолчанию — ``/data/awg-dash.db`` (персистентный volume контейнера),
переопределяется аргументом или переменной окружения ``AWG_DASH_DB``.
Каталог создаётся автоматически, поэтому на пустом ``/data`` схема v1
поднимается без ручных действий.

Соединение открывается в режиме WAL: ``synchronous=NORMAL`` достаточен для WAL
(риск потери только последних транзакций при падении операционной системы, БД
при этом не разрушается).

Контракт потокобезопасности (#54): соединение одно на процесс, все операции —
и записи (:meth:`Database.transaction`), и чтения (:meth:`Database.read`) —
выполняются под одним ``RLock``. Поэтому:

* результаты операторов из разных потоков не перемешиваются (нет
  ``TypeError``/``IndexError``);
* читатель не видит незафиксированных строк чужой транзакции и состояния,
  которое затем будет откачено (падение внешнего применителя);
* конкурентные запросы ждут, не падают.

WAL при этом остаётся: он даёт устойчивость к падению процесса и позволяет
читать БД снаружи (``sqlite3`` CLI) во время работы шлюза. Если позже
понадобится параллельная пропускная способность чтений, контракт меняется на
«соединение на поток» (``threading.local``) — это отдельное решение, не
молчаливое поведение.
"""

from __future__ import annotations

import os
import sqlite3
import threading
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path

from .errors import StorageError
from .migrations import MigrationReport, current_version, migrate, verify_schema

DEFAULT_DB_PATH = "/data/awg-dash.db"
DB_PATH_ENV = "AWG_DASH_DB"
BUSY_TIMEOUT_MS = 5000

_WAL_PRAGMAS: tuple[str, ...] = (
    "PRAGMA journal_mode = WAL",
    "PRAGMA synchronous = NORMAL",
    "PRAGMA foreign_keys = ON",
    "PRAGMA busy_timeout = 5000",
)


def resolve_db_path(path: str | os.PathLike[str] | None = None) -> Path:
    """Определить путь к БД: аргумент -> ``AWG_DASH_DB`` -> ``/data/awg-dash.db``."""
    if path is None:
        path = os.environ.get(DB_PATH_ENV) or DEFAULT_DB_PATH
    return Path(path)


class Database:
    """Соединение SQLite: миграции, проверка схемы, транзакции."""

    def __init__(
        self,
        path: str | os.PathLike[str] | None = None,
        *,
        auto_migrate: bool = True,
    ) -> None:
        self.path = resolve_db_path(path)
        self._lock = threading.RLock()
        self._conn = self._open(self.path)
        self.migration_report: MigrationReport | None = None
        if auto_migrate:
            self.migrate()
        else:
            verify_schema(self._conn)

    # -- соединение -----------------------------------------------------
    @staticmethod
    def _open(path: Path) -> sqlite3.Connection:
        if str(path) != ":memory:":
            try:
                path.parent.mkdir(parents=True, exist_ok=True)
            except OSError as exc:
                raise StorageError(f"не удалось создать каталог {path.parent}: {exc}") from exc
        try:
            conn = sqlite3.connect(
                str(path),
                isolation_level=None,
                check_same_thread=False,
                timeout=BUSY_TIMEOUT_MS / 1000,
            )
            conn.row_factory = sqlite3.Row
            for pragma in _WAL_PRAGMAS:
                conn.execute(pragma)
        except sqlite3.Error as exc:
            raise StorageError(f"не удалось открыть БД {path}: {exc}") from exc
        return conn

    @property
    def connection(self) -> sqlite3.Connection:
        """Прямой доступ к соединению — только для однопоточного кода.

        Рантайм API/сервисного слоя использует :meth:`read` (чтения) и
        :meth:`transaction` (мутации): они берут общий лок и гарантируют
        согласованность результата (#54).
        """
        return self._conn

    @property
    def journal_mode(self) -> str:
        with self._lock:
            row = self._conn.execute("PRAGMA journal_mode").fetchone()
        return str(row[0]) if row is not None else ""

    @property
    def is_wal(self) -> bool:
        return self.journal_mode.lower() == "wal"

    @property
    def schema_version(self) -> int:
        with self._lock:
            return current_version(self._conn)

    # -- схема ----------------------------------------------------------
    def migrate(self) -> MigrationReport:
        """Применить неприменённые миграции и проверить схему."""
        with self._lock:
            report = migrate(self._conn)
        self.migration_report = report
        return report

    def verify_schema(self) -> None:
        """Проверить соответствие фактической схемы ожидаемой версии."""
        with self._lock:
            verify_schema(self._conn)

    # -- транзакции и чтения --------------------------------------------
    @contextmanager
    def transaction(self) -> Iterator[sqlite3.Connection]:
        """Транзакция ALL-OR-NOTHING (``BEGIN IMMEDIATE`` + rollback при ошибке)."""
        with self._lock:
            self._conn.execute("BEGIN IMMEDIATE")
            try:
                yield self._conn
            except BaseException:
                self._conn.rollback()
                raise
            else:
                self._conn.commit()

    @contextmanager
    def read(self) -> Iterator[sqlite3.Connection]:
        """Согласованное чтение под общим локом (#54).

        Ждёт завершения чужой транзакции и потому не видит незафиксированных
        строк и состояния, которое затем будет откачено. Внутри уже открытой
        транзакции того же потока лок реентерабелен, поэтому читать можно и из
        неё.
        """
        with self._lock:
            yield self._conn

    def close(self) -> None:
        with self._lock:
            try:
                self._conn.close()
            except sqlite3.Error as exc:  # pragma: no cover - защита от гонок при shutdown
                raise StorageError(f"ошибка закрытия БД: {exc}") from exc

    def __enter__(self) -> Database:
        return self

    def __exit__(self, *exc_info: object) -> None:
        self.close()