from __future__ import annotations import hashlib import json import os import time from pathlib import Path from typing import Any def cache_root(config: dict) -> Path: root = Path(config.get("paths", {}).get("cache") or Path(config["paths"]["data"]) / "cache") root.mkdir(parents=True, exist_ok=True) return root def cache_path(config: dict, namespace: str, key: str) -> Path: digest = hashlib.sha256(key.encode()).hexdigest() path = cache_root(config) / namespace / f"{digest}.json" path.parent.mkdir(parents=True, exist_ok=True) return path def get_json(config: dict, namespace: str, key: str, ttl_seconds: int | None = None) -> Any | None: path = cache_path(config, namespace, key) if not path.exists(): return None if ttl_seconds is not None and time.time() - path.stat().st_mtime > ttl_seconds: return None try: return json.loads(path.read_text()) except (OSError, json.JSONDecodeError): return None def set_json(config: dict, namespace: str, key: str, value: Any) -> None: path = cache_path(config, namespace, key) tmp = path.with_suffix(".tmp") tmp.write_text(json.dumps(value, sort_keys=True)) tmp.replace(path) prune(config) def remove_json(config: dict, namespace: str, key: str) -> None: path = cache_path(config, namespace, key) try: path.unlink() except FileNotFoundError: return def prune(config: dict) -> None: root = cache_root(config) max_bytes = int(config.get("app", {}).get("cache_max_bytes", 20 * 1024**3)) files = [] total = 0 for current, _, names in os.walk(root): for name in names: path = Path(current) / name try: stat = path.stat() except OSError: continue total += stat.st_size files.append((stat.st_mtime, stat.st_size, path)) if total <= max_bytes: return for _, size, path in sorted(files): try: path.unlink() total -= size except OSError: continue if total <= max_bytes: break