| """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) |
|
|
