import json import os import tempfile import time from dataclasses import dataclass from pathlib import Path from typing import Any @dataclass class CacheEntry: value: Any stale: bool class FileCache: def __init__(self, cache_dir: str, default_ttl_seconds: int) -> None: self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(parents=True, exist_ok=True) self.default_ttl_seconds = default_ttl_seconds def _path_for(self, key: str) -> Path: safe_key = "".join(c if c.isalnum() or c in ("-", "_") else "_" for c in key) return self.cache_dir / f"{safe_key}.json" def _read_raw(self, key: str) -> dict[str, Any] | None: path = self._path_for(key) if not path.exists(): return None try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None def get_json(self, key: str, allow_stale: bool = False) -> CacheEntry | None: payload = self._read_raw(key) if payload is None: return None expires_at = float(payload.get("expires_at", 0)) value = payload.get("value") stale = time.time() > expires_at if stale and not allow_stale: return None return CacheEntry(value=value, stale=stale) def set_json(self, key: str, value: Any, ttl_seconds: int | None = None) -> None: ttl = ttl_seconds if ttl_seconds is not None else self.default_ttl_seconds payload = { "expires_at": time.time() + max(ttl, 0), "value": value, } self._atomic_write_json(self._path_for(key), payload) def _atomic_write_json(self, path: Path, payload: dict[str, Any]) -> None: fd, tmp_name = tempfile.mkstemp(prefix="cache_", suffix=".tmp", dir=str(self.cache_dir)) try: with os.fdopen(fd, "w", encoding="utf-8") as tmp_file: json.dump(payload, tmp_file, separators=(",", ":")) os.replace(tmp_name, path) finally: if os.path.exists(tmp_name): os.remove(tmp_name)