57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from threading import RLock
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SHARED_USER = "shared"
|
|
MA_ALERT_PERIODS = (10, 20, 50, 100, 200)
|
|
|
|
|
|
class UserPrefStore:
|
|
"""Per-user JSON bags, one file.
|
|
|
|
Namespaces are versioned objects. ``ma_alerts`` is ``{"1d": [200]}``.
|
|
``user_id`` is ``shared`` until OIDC supplies a subject.
|
|
"""
|
|
|
|
def __init__(self, path: str | Path):
|
|
self.path = Path(path)
|
|
self._lock = RLock()
|
|
self._by_user: dict[str, dict[str, Any]] = self._load()
|
|
|
|
def _load(self) -> dict[str, dict[str, Any]]:
|
|
if not self.path.exists():
|
|
return {}
|
|
try:
|
|
payload = json.loads(self.path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
logger.warning("Could not read user prefs; starting empty", exc_info=True)
|
|
return {}
|
|
if not isinstance(payload, dict):
|
|
return {}
|
|
loaded: dict[str, dict[str, Any]] = {}
|
|
for user_id, namespaces in payload.items():
|
|
if isinstance(namespaces, dict):
|
|
loaded[str(user_id)] = namespaces
|
|
return loaded
|
|
|
|
def _save(self) -> None:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = self.path.with_suffix(self.path.suffix + ".tmp")
|
|
temporary.write_text(
|
|
json.dumps(self._by_user, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
temporary.replace(self.path)
|
|
|
|
def get(self, namespace: str, default: Any = None, user_id: str = SHARED_USER) -> Any:
|
|
return self._by_user.get(user_id, {}).get(namespace, default)
|
|
|
|
def put(self, namespace: str, value: Any, user_id: str = SHARED_USER) -> Any:
|
|
with self._lock:
|
|
self._by_user.setdefault(user_id, {})[namespace] = value
|
|
self._save()
|
|
return value
|