"""Content-addressed cache that includes all factors affecting a component output.""" from __future__ import annotations import json import uuid from pathlib import Path from typing import Any from .ids import stable_id class JsonCache: def __init__(self, root: Path, enabled: bool = True) -> None: self.root = root self.enabled = enabled if enabled: root.mkdir(parents=True, exist_ok=True) @staticmethod def key(component: str, payload: dict[str, Any]) -> str: return stable_id(f"cache:{component}", payload) def get(self, component: str, payload: dict[str, Any]) -> dict[str, Any] | None: if not self.enabled: return None location = self.root / component / f"{self.key(component, payload)}.json" if not location.is_file(): return None return json.loads(location.read_text(encoding="utf-8")) def put(self, component: str, payload: dict[str, Any], value: dict[str, Any]) -> None: if not self.enabled: return location = self.root / component / f"{self.key(component, payload)}.json" location.parent.mkdir(parents=True, exist_ok=True) temporary = location.with_suffix(f".{uuid.uuid4().hex}.tmp") temporary.write_text(json.dumps(value, ensure_ascii=False, sort_keys=True), encoding="utf-8") temporary.replace(location)