"""Locked, atomically replaced JSON files for the small per-user registries.""" from __future__ import annotations import fcntl import json import os from pathlib import Path import tempfile from typing import Callable, Generic, TypeVar Value = TypeVar("Value") Outcome = TypeVar("Outcome") class JsonRegistry(Generic[Value]): """A private JSON file updated under an exclusive lock. Subclasses provide the user-facing error type and the mapping between the stored payload and a Python value. Every update reads, modifies and writes under the lock so concurrent Hub processes cannot lose each other's changes, the file is replaced atomically with mode 0600, and a corrupt file is never overwritten. """ error_type: type[RuntimeError] = RuntimeError def __init__(self, path: Path) -> None: self.path = path @property def lock_path(self) -> Path: return self.path.with_name(f".{self.path.name}.lock") def _empty(self) -> Value: raise NotImplementedError def _parse(self, payload: object) -> Value: raise NotImplementedError def _serialize(self, value: Value) -> object: raise NotImplementedError def load(self) -> Value: if not self.path.exists(): return self._empty() try: payload = json.loads(self.path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise self.error_type(f"cannot read {self.path}: {error}") from error return self._parse(payload) def _update(self, callback: Callable[[Value], tuple[Value, Outcome]]) -> Outcome: try: self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) with self.lock_path.open("a+", encoding="utf-8") as lock: fcntl.flock(lock.fileno(), fcntl.LOCK_EX) current = self.load() updated, outcome = callback(current) if updated != current: self._write(updated) return outcome except self.error_type: raise except OSError as error: raise self.error_type(f"cannot update {self.path}: {error}") from error def _write(self, value: Value) -> None: payload = self._serialize(value) temporary_name: str | None = None try: with tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", dir=self.path.parent, prefix=f".{self.path.name}.", delete=False, ) as temporary: temporary_name = temporary.name json.dump(payload, temporary, ensure_ascii=False, indent=2) temporary.write("\n") temporary.flush() os.fsync(temporary.fileno()) os.chmod(temporary_name, 0o600) os.replace(temporary_name, self.path) directory_fd = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(directory_fd) finally: os.close(directory_fd) finally: if temporary_name is not None: try: Path(temporary_name).unlink(missing_ok=True) except OSError: pass