| """Small atomic JSON store for a single-process LibBee deployment.""" |
|
|
| import copy |
| import json |
| import os |
| import tempfile |
| import time |
| from pathlib import Path |
| from typing import Any, Dict, Optional |
|
|
|
|
| class JsonRuntimeStore: |
| def __init__(self, path: Path, default: Optional[Dict[str, Any]] = None): |
| self.path = Path(path) |
| self.path.parent.mkdir(parents=True, exist_ok=True) |
| self.default = copy.deepcopy(default or {}) |
|
|
| def load(self) -> Dict[str, Any]: |
| data = copy.deepcopy(self.default) |
| try: |
| loaded = json.loads(self.path.read_text(encoding="utf-8")) |
| if isinstance(loaded, dict): |
| data.update(loaded) |
| except (OSError, UnicodeError, json.JSONDecodeError): |
| pass |
| return data |
|
|
| def save(self, data: Dict[str, Any]) -> None: |
| """Replace a complete object with a unique, owner-only temporary file. |
| |
| A shared fixed `.tmp` name allowed independent writers to overwrite or |
| unlink each other's temporary file. Unique files prevent that race; |
| callers still need a database for multi-process read/modify/write use. |
| """ |
| if not isinstance(data, dict): |
| raise TypeError("runtime data must be a JSON object") |
| tmp = None |
| try: |
| with tempfile.NamedTemporaryFile( |
| mode="w", encoding="utf-8", dir=self.path.parent, |
| prefix=f".{self.path.name}.", suffix=".tmp", delete=False, |
| ) as handle: |
| tmp = Path(handle.name) |
| json.dump(data, handle, ensure_ascii=False, indent=2, allow_nan=False) |
| handle.flush() |
| os.fsync(handle.fileno()) |
| os.replace(tmp, self.path) |
| finally: |
| if tmp is not None: |
| tmp.unlink(missing_ok=True) |
|
|
| def update(self, **kwargs: Any) -> Dict[str, Any]: |
| data = self.load() |
| data.update(kwargs) |
| data["updated_at"] = time.time() |
| self.save(data) |
| return data |
|
|
