feat: preferences LocalFileStore — YAML-backed single-user preference store

This commit is contained in:
pyr0ball 2026-04-04 18:07:35 -07:00
parent 9ee31a09c1
commit 0d9d030320
2 changed files with 119 additions and 0 deletions

View file

@ -0,0 +1,75 @@
"""Preference store backends.
``LocalFileStore`` reads and writes a single YAML file at a configurable
path (default: ``~/.config/circuitforge/preferences.yaml``).
The ``PreferenceStore`` protocol describes the interface any backend must
satisfy. The Heimdall cloud backend will implement the same protocol once
Heimdall#5 (user_preferences column) lands — products swap backends by
passing a different store instance.
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Any, Protocol, runtime_checkable
from .paths import get_path, set_path
logger = logging.getLogger(__name__)
_DEFAULT_PREFS_PATH = Path.home() / ".config" / "circuitforge" / "preferences.yaml"
@runtime_checkable
class PreferenceStore(Protocol):
"""Read/write interface for user preferences.
``user_id`` is passed through for cloud backends that store per-user
data. Local single-user backends accept it but ignore it.
"""
def get(self, user_id: str | None, path: str, default: Any = None) -> Any:
"""Return the value at *path*, or *default* if missing."""
...
def set(self, user_id: str | None, path: str, value: Any) -> None:
"""Persist *value* at *path*."""
...
class LocalFileStore:
"""Single-user preference store backed by a YAML file.
Thread-safe for typical single-process use (reads the file on every
``get`` call, writes atomically via a temp-file rename on ``set``).
Not suitable for concurrent multi-process writes.
"""
def __init__(self, prefs_path: Path = _DEFAULT_PREFS_PATH) -> None:
self._path = Path(prefs_path)
def _load(self) -> dict:
if not self._path.exists():
return {}
try:
import yaml # type: ignore[import]
text = self._path.read_text(encoding="utf-8")
data = yaml.safe_load(text)
return data if isinstance(data, dict) else {}
except Exception as exc:
logger.warning("preferences: could not read %s: %s", self._path, exc)
return {}
def _save(self, data: dict) -> None:
import yaml # type: ignore[import]
self._path.parent.mkdir(parents=True, exist_ok=True)
tmp = self._path.with_suffix(".yaml.tmp")
tmp.write_text(yaml.safe_dump(data, default_flow_style=False), encoding="utf-8")
tmp.replace(self._path)
def get(self, user_id: str | None, path: str, default: Any = None) -> Any: # noqa: ARG002
return get_path(self._load(), path, default=default)
def set(self, user_id: str | None, path: str, value: Any) -> None: # noqa: ARG002
self._save(set_path(self._load(), path, value))

View file

@ -73,3 +73,47 @@ class TestSetPath:
prefs = set_path(prefs, "affiliate.byok_ids.ebay", "tag-123")
assert get_path(prefs, "affiliate.opt_out") is True
assert get_path(prefs, "affiliate.byok_ids.ebay") == "tag-123"
import os
import tempfile
from pathlib import Path
from circuitforge_core.preferences.store import LocalFileStore
class TestLocalFileStore:
def _store(self, tmp_path) -> LocalFileStore:
return LocalFileStore(prefs_path=tmp_path / "preferences.yaml")
def test_get_returns_default_when_file_missing(self, tmp_path):
store = self._store(tmp_path)
assert store.get(user_id=None, path="affiliate.opt_out", default=False) is False
def test_set_then_get_roundtrip(self, tmp_path):
store = self._store(tmp_path)
store.set(user_id=None, path="affiliate.opt_out", value=True)
assert store.get(user_id=None, path="affiliate.opt_out", default=False) is True
def test_set_nested_path(self, tmp_path):
store = self._store(tmp_path)
store.set(user_id=None, path="affiliate.byok_ids.ebay", value="my-tag")
assert store.get(user_id=None, path="affiliate.byok_ids.ebay") == "my-tag"
def test_set_preserves_sibling_keys(self, tmp_path):
store = self._store(tmp_path)
store.set(user_id=None, path="affiliate.opt_out", value=False)
store.set(user_id=None, path="affiliate.byok_ids.ebay", value="tag")
assert store.get(user_id=None, path="affiliate.opt_out") is False
assert store.get(user_id=None, path="affiliate.byok_ids.ebay") == "tag"
def test_creates_parent_dirs(self, tmp_path):
deep_path = tmp_path / "deep" / "nested" / "preferences.yaml"
store = LocalFileStore(prefs_path=deep_path)
store.set(user_id=None, path="x", value=1)
assert deep_path.exists()
def test_user_id_ignored_for_local_store(self, tmp_path):
"""LocalFileStore is single-user; user_id is accepted but ignored."""
store = self._store(tmp_path)
store.set(user_id="u123", path="affiliate.opt_out", value=True)
assert store.get(user_id="u456", path="affiliate.opt_out", default=False) is True