39 lines
1.1 KiB
Python
39 lines
1.1 KiB
Python
import json
|
|
from typing import Any
|
|
|
|
import aiosqlite
|
|
|
|
DEFAULTS: dict[str, Any] = {
|
|
"actions": True,
|
|
"commands": False,
|
|
"confirm": True,
|
|
"muted": ["ScheduleRun"],
|
|
"tz": "UTC",
|
|
"digest": None,
|
|
"digest_sent": "",
|
|
}
|
|
|
|
|
|
class SettingRepository:
|
|
def __init__(self, conn: aiosqlite.Connection) -> None:
|
|
self._conn = conn
|
|
self._cache: dict[str, Any] = {}
|
|
|
|
async def load(self) -> None:
|
|
cursor = await self._conn.execute("SELECT key, value FROM settings")
|
|
self._cache = {
|
|
row["key"]: json.loads(row["value"]) for row in await cursor.fetchall()
|
|
}
|
|
|
|
def get(self, key: str) -> Any: # noqa: ANN401
|
|
return self._cache.get(key, DEFAULTS[key])
|
|
|
|
async def set(self, key: str, value: Any) -> None: # noqa: ANN401
|
|
self._cache[key] = value
|
|
await self._conn.execute(
|
|
"INSERT INTO settings (key, value) VALUES (?, ?) "
|
|
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
|
(key, json.dumps(value, ensure_ascii=False)),
|
|
)
|
|
await self._conn.commit()
|