refactor: extract TableMigrations and DbStore base class
lint / lint (push) Successful in 2m0s
test / unit (pull_request) Successful in 1m4s
test / integration (pull_request) Successful in 23s
test / coverage (pull_request) Successful in 1m9s

Adds bot_bottle/migrations.py (TableMigrations) and bot_bottle/db_store.py
(DbStore) per PR review. Both stores now inherit from DbStore and hold a
TableMigrations instance instead of duplicating schema-version logic inline.
This commit is contained in:
2026-07-02 21:04:34 +00:00
parent e8e4f6f7c7
commit a4d82e5ff2
5 changed files with 110 additions and 91 deletions
+15 -45
View File
@@ -9,6 +9,13 @@ from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .supervise import AuditEntry
try:
from .db_store import DbStore
from .migrations import TableMigrations
except ImportError:
from db_store import DbStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from migrations import TableMigrations # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
def get_supervise_mod() -> object:
"""Lazy import of supervise to avoid a circular-import at module init time.
@@ -26,10 +33,10 @@ def get_supervise_mod() -> object:
return _m
# One entry per schema version: _MIGRATIONS[0] brings a fresh DB (user_version=0)
# to version 1, _MIGRATIONS[1] to version 2, and so on. Add new migrations at
# the end; never edit existing ones.
_MIGRATIONS: list[str] = [
# One entry per schema version: _MIGRATIONS.migrations[0] brings a fresh DB
# to version 1, [1] to version 2, and so on. Add new migrations at the end;
# never edit existing ones.
_MIGRATIONS = TableMigrations("audit_store", [
# v1 — initial schema
"""
CREATE TABLE IF NOT EXISTS supervise_audit_entries (
@@ -43,16 +50,15 @@ _MIGRATIONS: list[str] = [
diff TEXT NOT NULL
)
""",
]
])
class AuditStore:
class AuditStore(DbStore):
"""SQLite-backed persistent store for supervise audit entries."""
def __init__(self, db_path: Path | None = None) -> None:
self.db_path = db_path or get_supervise_mod().host_db_path() # type: ignore[attr-defined]
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self._init()
resolved = db_path or get_supervise_mod().host_db_path() # type: ignore[attr-defined]
super().__init__(resolved, _MIGRATIONS)
def write_audit_entry(self, entry: AuditEntry) -> Path:
with self._connect() as conn:
@@ -103,41 +109,5 @@ class AuditStore:
diff=row["diff"],
)
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
_SCHEMA_KEY = "audit_store"
def _init(self) -> None:
with self._connect() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS schema_versions (
module TEXT PRIMARY KEY,
version INTEGER NOT NULL DEFAULT 0
)
"""
)
row = conn.execute(
"SELECT version FROM schema_versions WHERE module = ?",
(self._SCHEMA_KEY,),
).fetchone()
version = row[0] if row else 0
for i, sql in enumerate(_MIGRATIONS[version:], start=version + 1):
conn.execute(sql)
conn.execute(
"INSERT OR REPLACE INTO schema_versions (module, version) VALUES (?, ?)",
(self._SCHEMA_KEY, i),
)
self._chmod()
def _chmod(self) -> None:
try:
self.db_path.chmod(0o600)
except OSError:
pass
__all__ = ["AuditStore"]