didericis-claude
142974a4b8
test / unit (pull_request) Successful in 58s
test / integration (pull_request) Successful in 18s
test / coverage (pull_request) Successful in 1m6s
lint / lint (push) Successful in 1m55s
prd-number / assign-numbers (push) Successful in 16s
test / unit (push) Successful in 54s
test / integration (push) Successful in 19s
test / coverage (push) Successful in 1m6s
Update Quality Badges / update-badges (push) Successful in 1m2s
Boolean-returning methods should read as predicates. Renames DbStore.check_migrations, StoreManager.check_migrations, and the test patch accordingly.
61 lines
2.0 KiB
Python
61 lines
2.0 KiB
Python
"""Singleton manager for all bot-bottle SQLite stores (PRD 0013)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from .audit_store import AuditStore
|
|
from .queue_store import QueueStore
|
|
except ImportError:
|
|
from audit_store import AuditStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
|
|
from queue_store import QueueStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
|
|
|
|
_instance: StoreManager | None = None
|
|
|
|
|
|
class StoreManager:
|
|
"""Owns db_path and delegates migrate/is_migrated across all stores.
|
|
|
|
Use instance() for normal access. Call reset(db_path) in tests to swap
|
|
the singleton to a temp path, then reset() with no args to restore the
|
|
default."""
|
|
|
|
def __init__(self, db_path: Path | None = None) -> None:
|
|
if db_path is None:
|
|
# Lazy import to avoid a circular dependency: supervise imports
|
|
# StoreManager at module level; StoreManager must not import
|
|
# supervise at module level in return.
|
|
try:
|
|
from .supervise import host_db_path
|
|
except ImportError:
|
|
from supervise import host_db_path # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
|
|
db_path = host_db_path()
|
|
self.db_path = db_path
|
|
|
|
@classmethod
|
|
def instance(cls) -> StoreManager:
|
|
global _instance
|
|
if _instance is None:
|
|
_instance = cls()
|
|
return _instance
|
|
|
|
@classmethod
|
|
def reset(cls, db_path: Path | None = None) -> None:
|
|
"""Replace the singleton. Pass db_path for test isolation; omit to restore default."""
|
|
global _instance
|
|
_instance = cls(db_path)
|
|
|
|
def is_migrated(self) -> bool:
|
|
return (
|
|
QueueStore("", self.db_path).is_migrated()
|
|
and AuditStore(self.db_path).is_migrated()
|
|
)
|
|
|
|
def migrate(self) -> None:
|
|
QueueStore("", self.db_path).migrate()
|
|
AuditStore(self.db_path).migrate()
|
|
|
|
|
|
__all__ = ["StoreManager"]
|