didericis-claude
5fcca2060f
Remove auto-migration from DbStore.__init__; add explicit check_migrations() and migrate() methods. The CLI now checks both stores on every startup and prompts the user to confirm before migrating. supervise.prepare() calls .migrate() directly now that __init__ no longer does it implicitly. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
"""Shared SQLite-backed store base class for bot-bottle (PRD 0013)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from .migrations import TableMigrations
|
|
except ImportError:
|
|
from migrations import TableMigrations # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
|
|
|
|
|
|
class DbVersionError(Exception):
|
|
"""Raised when the on-disk schema is behind the current migration list."""
|
|
|
|
|
|
class DbStore:
|
|
"""Base for SQLite-backed stores. Subclasses resolve db_path then call super().__init__."""
|
|
|
|
def __init__(self, db_path: Path, migrations: TableMigrations) -> None:
|
|
self.db_path = db_path
|
|
self._migrations = migrations
|
|
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(self.db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def check_migrations(self) -> bool:
|
|
"""Return True if the DB is fully up-to-date, False if migration is needed."""
|
|
if not self.db_path.exists():
|
|
return False
|
|
try:
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT version FROM schema_versions WHERE module = ?",
|
|
(self._migrations.schema_key,),
|
|
).fetchone()
|
|
except sqlite3.OperationalError:
|
|
return False
|
|
version = row[0] if row else 0
|
|
return version == len(self._migrations.migrations)
|
|
|
|
def migrate(self) -> None:
|
|
"""Apply any pending migrations and set permissions on the DB file."""
|
|
with self._connect() as conn:
|
|
self._migrations.apply(conn)
|
|
self._chmod()
|
|
|
|
def _chmod(self) -> None:
|
|
try:
|
|
self.db_path.chmod(0o600)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
__all__ = ["DbStore", "DbVersionError"]
|