File size: 1,708 Bytes
37ae25d | 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 | """
runtime_store.py — LibBee v3.1
Fixes applied:
1. save() now uses atomic write-then-rename (os.replace) to prevent file
corruption on crash or mid-write failure.
os.replace is atomic on POSIX (Linux/macOS) and effectively atomic on
Windows (same filesystem). This protects feedback.jsonl, metrics.json,
and runtime_config.json from partial writes.
"""
import json
import os
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 = default or {}
def load(self) -> Dict[str, Any]:
if not self.path.exists():
return dict(self.default)
try:
return json.loads(self.path.read_text(encoding="utf-8"))
except Exception:
return dict(self.default)
def save(self, data: Dict[str, Any]) -> None:
"""Atomically write data to disk using a temp file + os.replace."""
tmp = self.path.with_suffix(".tmp")
try:
tmp.write_text(
json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
)
os.replace(tmp, self.path)
except Exception:
try:
tmp.unlink(missing_ok=True)
except Exception:
pass
raise
def update(self, **kwargs: Any) -> Dict[str, Any]:
data = self.load()
data.update(kwargs)
data["updated_at"] = time.time()
self.save(data)
return data
|