jmri_server.roster_store

  1"""``RosterStore`` — server-owned roster persistence.
  2
  3Backs the roster service. Holds an in-memory dict of ``RosterEntry`` and
  4persists each entry as its own ``locomotive-config`` XML file under
  5``Profile.roster_dir`` (the Stage 2 directory layout).
  6
  7Storage policy:
  8
  9* One file per entry, named ``<id>.xml``.
 10* The roster index ``roster.xml`` is regenerated from the in-memory
 11  state on every mutation.
 12* In-memory state is the source of truth during a server session; the
 13  XML files are the source of truth at startup.
 14"""
 15from __future__ import annotations
 16
 17import logging
 18import re
 19from collections.abc import Iterator
 20from dataclasses import dataclass, field
 21
 22from jmri_core.beans.roster_entry import RosterEntry
 23from jmri_core.persistence import (
 24    RosterIndex,
 25    load_locomotive_bytes,
 26    save_locomotive,
 27    save_roster_index,
 28)
 29from jmri_core.persistence.profile import Profile
 30
 31_log = logging.getLogger(__name__)
 32_SAFE_FILENAME = re.compile(r"[^A-Za-z0-9._-]+")
 33
 34
 35def _safe_filename(entry_id: str) -> str:
 36    """Sanitize an entry id into a filesystem-safe filename stem."""
 37    cleaned = _SAFE_FILENAME.sub("_", entry_id).strip("_") or "entry"
 38    return f"{cleaned}.xml"
 39
 40
 41@dataclass
 42class RosterStore:
 43    """In-memory + on-disk roster."""
 44
 45    profile: Profile
 46    _entries: dict[str, RosterEntry] = field(default_factory=dict)
 47
 48    def __post_init__(self) -> None:
 49        self.profile.ensure()
 50        self._load_existing()
 51
 52    def _load_existing(self) -> None:
 53        """Read every ``locomotive-config`` XML in roster_dir into memory."""
 54        roster_dir = self.profile.roster_dir
 55        if not roster_dir.is_dir():
 56            return
 57        for path in roster_dir.glob("*.xml"):
 58            if path.name == "roster.xml":
 59                continue
 60            try:
 61                entries = load_locomotive_bytes(path.read_bytes())
 62            except Exception as e:  # noqa: BLE001
 63                _log.warning("RosterStore: failed to load %s: %s", path, e)
 64                continue
 65            for entry in entries:
 66                if entry.id:
 67                    self._entries[entry.id] = entry
 68
 69    # ----- read --------------------------------------------------------
 70
 71    def list(self) -> list[RosterEntry]:
 72        return list(self._entries.values())
 73
 74    def get(self, entry_id: str) -> RosterEntry | None:
 75        return self._entries.get(entry_id)
 76
 77    def __iter__(self) -> Iterator[RosterEntry]:
 78        return iter(self._entries.values())
 79
 80    def __len__(self) -> int:
 81        return len(self._entries)
 82
 83    # ----- write -------------------------------------------------------
 84
 85    def put(self, entry: RosterEntry) -> RosterEntry:
 86        """Create-or-replace by id. Persists to disk + rewrites the index."""
 87        if not entry.id:
 88            raise ValueError("RosterEntry.id is required")
 89        self._entries[entry.id] = entry
 90        path = self.profile.roster_dir / _safe_filename(entry.id)
 91        save_locomotive(path, [entry])
 92        # Update the file_name field so reloads find it.
 93        if not entry.file_name:
 94            entry.file_name = path.name
 95            save_locomotive(path, [entry])
 96        self._write_index()
 97        return entry
 98
 99    def delete(self, entry_id: str) -> bool:
100        """Remove an entry. Returns True if it existed."""
101        if entry_id not in self._entries:
102            return False
103        del self._entries[entry_id]
104        path = self.profile.roster_dir / _safe_filename(entry_id)
105        if path.exists():
106            try:
107                path.unlink()
108            except OSError as e:
109                _log.warning("RosterStore: failed to delete %s: %s", path, e)
110        self._write_index()
111        return True
112
113    # ----- index ------------------------------------------------------
114
115    def _write_index(self) -> None:
116        index = RosterIndex(entries=list(self._entries.values()))
117        try:
118            save_roster_index(self.profile.roster_index, index)
119        except OSError as e:
120            _log.warning("RosterStore: failed to write roster index: %s", e)