fix: make config store schema explicit
This commit is contained in:
+23
-31
@@ -14,66 +14,58 @@ except ImportError:
|
||||
from supervise_types import host_db_path # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
|
||||
|
||||
|
||||
CACHED_IMAGE_STALE_WARNING_DAYS = "cached_image_stale_warning_days"
|
||||
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS = 1
|
||||
|
||||
|
||||
class ConfigStore(DbStore):
|
||||
"""SQLite key/value configuration for host-side bot-bottle settings."""
|
||||
"""SQLite configuration for host-side bot-bottle settings."""
|
||||
|
||||
def __init__(self, db_path: Path | None = None) -> None:
|
||||
migrations = TableMigrations("config_store", [
|
||||
# v1 — generic string key/value settings
|
||||
# v1 — host-side bot-bottle settings
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS bot_bottle_config (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
cached_image_stale_warning_days INTEGER NOT NULL DEFAULT 1
|
||||
)
|
||||
""",
|
||||
])
|
||||
super().__init__(db_path or host_db_path(), migrations)
|
||||
|
||||
def get(self, key: str) -> str | None:
|
||||
def cached_image_stale_warning_days(self) -> int:
|
||||
if not self.db_path.is_file():
|
||||
return None
|
||||
return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
|
||||
with self._connect() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT value FROM bot_bottle_config WHERE key = ?",
|
||||
(key,),
|
||||
"""
|
||||
SELECT cached_image_stale_warning_days
|
||||
FROM bot_bottle_config
|
||||
WHERE id = 1
|
||||
""",
|
||||
).fetchone()
|
||||
return str(row["value"]) if row is not None else None
|
||||
if row is None:
|
||||
return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
|
||||
try:
|
||||
return int(row["cached_image_stale_warning_days"])
|
||||
except (TypeError, ValueError):
|
||||
return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
|
||||
|
||||
def set(self, key: str, value: str) -> Path:
|
||||
def set_cached_image_stale_warning_days(self, days: int) -> Path:
|
||||
with self._connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO bot_bottle_config (key, value)
|
||||
VALUES (?, ?)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value
|
||||
INSERT INTO bot_bottle_config (id, cached_image_stale_warning_days)
|
||||
VALUES (1, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
cached_image_stale_warning_days = excluded.cached_image_stale_warning_days
|
||||
""",
|
||||
(key, value),
|
||||
(days,),
|
||||
)
|
||||
self._chmod()
|
||||
return self.db_path
|
||||
|
||||
def get_int(self, key: str, default: int) -> int:
|
||||
raw = self.get(key)
|
||||
if raw is None:
|
||||
return default
|
||||
try:
|
||||
return int(raw)
|
||||
except ValueError:
|
||||
return default
|
||||
|
||||
def cached_image_stale_warning_days(self) -> int:
|
||||
return self.get_int(
|
||||
CACHED_IMAGE_STALE_WARNING_DAYS,
|
||||
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"CACHED_IMAGE_STALE_WARNING_DAYS",
|
||||
"DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS",
|
||||
"ConfigStore",
|
||||
]
|
||||
|
||||
@@ -2,12 +2,12 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from bot_bottle.config_store import (
|
||||
CACHED_IMAGE_STALE_WARNING_DAYS,
|
||||
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
|
||||
ConfigStore,
|
||||
)
|
||||
@@ -28,9 +28,24 @@ class TestConfigStore(unittest.TestCase):
|
||||
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
|
||||
store = ConfigStore(Path(tmp) / "bot-bottle.db")
|
||||
store.migrate()
|
||||
store.set(CACHED_IMAGE_STALE_WARNING_DAYS, "7")
|
||||
store.set_cached_image_stale_warning_days(7)
|
||||
self.assertEqual(7, store.cached_image_stale_warning_days())
|
||||
|
||||
def test_config_schema_uses_explicit_settings_columns(self) -> None:
|
||||
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
|
||||
store = ConfigStore(Path(tmp) / "bot-bottle.db")
|
||||
store.migrate()
|
||||
with sqlite3.connect(store.db_path) as conn:
|
||||
conn.row_factory = sqlite3.Row
|
||||
columns = [
|
||||
row["name"]
|
||||
for row in conn.execute("PRAGMA table_info(bot_bottle_config)")
|
||||
]
|
||||
self.assertEqual([
|
||||
"id",
|
||||
"cached_image_stale_warning_days",
|
||||
], columns)
|
||||
|
||||
def test_store_manager_includes_config_store(self) -> None:
|
||||
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
|
||||
db = Path(tmp) / "bot-bottle.db"
|
||||
|
||||
Reference in New Issue
Block a user