102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
"""Singleton store for service groups with JSON file persistence.
|
|
|
|
Groups are persisted to a JSON file specified by the ``GROUPS_PATH``
|
|
environment variable (default ``/app/groups.json``).
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import threading
|
|
|
|
from server.models import Group, GroupsData
|
|
|
|
|
|
class GroupStore:
|
|
"""Thread-safe singleton that manages groups and persists to disk."""
|
|
|
|
_instance: "GroupStore | None" = None
|
|
_lock = threading.Lock()
|
|
|
|
def __new__(cls) -> "GroupStore":
|
|
with cls._lock:
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
cls._instance._initialized = False
|
|
return cls._instance
|
|
|
|
def __init__(self) -> None:
|
|
if self._initialized:
|
|
return
|
|
self._initialized = True
|
|
self._data_lock = threading.Lock()
|
|
self._path = os.environ.get("GROUPS_PATH", "/app/groups.json")
|
|
self._groups: list[Group] = []
|
|
self._load()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Persistence
|
|
# ------------------------------------------------------------------
|
|
|
|
def _load(self) -> None:
|
|
"""Load groups from the JSON file on disk."""
|
|
try:
|
|
with open(self._path) as f:
|
|
raw = json.load(f)
|
|
data = GroupsData(**raw)
|
|
self._groups = data.groups
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
self._groups = []
|
|
|
|
def _save(self) -> None:
|
|
"""Persist the current groups list to the JSON file."""
|
|
data = GroupsData(groups=self._groups)
|
|
with open(self._path, "w") as f:
|
|
json.dump(data.model_dump(), f, indent=2)
|
|
|
|
# ------------------------------------------------------------------
|
|
# CRUD operations
|
|
# ------------------------------------------------------------------
|
|
|
|
def list_groups(self) -> list[Group]:
|
|
"""Return all groups."""
|
|
with self._data_lock:
|
|
return list(self._groups)
|
|
|
|
def get_group(self, group_id: str) -> Group | None:
|
|
"""Return a single group by id, or None."""
|
|
with self._data_lock:
|
|
for g in self._groups:
|
|
if g.id == group_id:
|
|
return g
|
|
return None
|
|
|
|
def create_group(self, group: Group) -> Group:
|
|
"""Add a new group. Raises ValueError if duplicate id."""
|
|
with self._data_lock:
|
|
for g in self._groups:
|
|
if g.id == group.id:
|
|
raise ValueError(f"Group '{group.id}' already exists")
|
|
self._groups.append(group)
|
|
self._save()
|
|
return group
|
|
|
|
def update_group(self, group_id: str, group: Group) -> Group | None:
|
|
"""Replace a group by id. Returns updated group or None."""
|
|
with self._data_lock:
|
|
for i, g in enumerate(self._groups):
|
|
if g.id == group_id:
|
|
self._groups[i] = group
|
|
self._save()
|
|
return group
|
|
return None
|
|
|
|
def delete_group(self, group_id: str) -> bool:
|
|
"""Remove a group by id. Returns True if found and deleted."""
|
|
with self._data_lock:
|
|
for i, g in enumerate(self._groups):
|
|
if g.id == group_id:
|
|
self._groups.pop(i)
|
|
self._save()
|
|
return True
|
|
return False
|