File size: 2,040 Bytes
a2416df | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | """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
|