didericis
d7a58e52fd
lint / lint (push) Successful in 54s
test / integration-docker (pull_request) Successful in 38s
test / unit (pull_request) Successful in 45s
test / integration-firecracker (pull_request) Successful in 3m31s
test / coverage (pull_request) Successful in 18s
test / publish-infra (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Failing after 6s
Move the DbStore base (db_store), the shared schema-migration base (migrations / TableMigrations), the concrete stores (audit_store, config_store, queue_store), and StoreManager out of the package root into a bot_bottle/store/ package, so persistence lives in one place rather than scattered across the root. Callers use direct submodule imports (from bot_bottle.store.store_manager import StoreManager). Inside the moved modules the relative imports point back up to their non-store siblings (..paths, ..supervise_types) while co-moved siblings stay package-local; the flat-import fallbacks are untouched. The orchestrator's own stores (config_store, secret_store, registry) stay in orchestrator/ and repoint to the moved DbStore / TableMigrations. No behavior change; full unit suite green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
62 lines
2.1 KiB
Python
62 lines
2.1 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 .config_store import ConfigStore
|
|
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 config_store import ConfigStore # 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:
|
|
try:
|
|
from ..paths import host_db_path
|
|
except ImportError:
|
|
from paths 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()
|
|
and ConfigStore(self.db_path).is_migrated()
|
|
)
|
|
|
|
def migrate(self) -> None:
|
|
QueueStore("", self.db_path).migrate()
|
|
AuditStore(self.db_path).migrate()
|
|
ConfigStore(self.db_path).migrate()
|
|
|
|
|
|
__all__ = ["StoreManager"]
|