"""Открытие и жизненный цикл 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()