didericis
421d31c32f
Create bot_bottle/paths.py as the canonical home for the app-root path helpers (bot_bottle_root, host_db_path, HOST_DB_FILENAME) — foundational, not supervise- or db-specific. `bot_bottle_root()` now honours a BOT_BOTTLE_ROOT env override. Repoint every consumer (supervise, supervise_types, db_store, queue_store, audit_store, store_manager, bottle_state, cli/supervise, docker/cleanup, orchestrator/registry) at paths; remove the definitions (and supervise's duplicate host_db_path) and the now-dead `import sys`. Add paths.py to the sidecar bundle (Dockerfile.sidecars) for the flat-import copies. Tests: replace ~12 files' monkeypatching of supervise.bot_bottle_root (and the flat/pkg/supervise_types triple-patch dance) with a single `use_bottle_root()` helper that sets BOT_BOTTLE_ROOT — every module and flat/package copy reads the same env var, so one override covers them all. Net -97 lines. Behaviour-preserving: full unit suite unchanged (only the pre-existing /bin/sleep sidecar-init errors remain). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
58 lines
1.8 KiB
Python
58 lines
1.8 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:
|
|
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()
|
|
)
|
|
|
|
def migrate(self) -> None:
|
|
QueueStore("", self.db_path).migrate()
|
|
AuditStore(self.db_path).migrate()
|
|
|
|
|
|
__all__ = ["StoreManager"]
|