src/veil_pgd/io/cache.py

"""Simple on-disk JSON KV cache keyed by a hash of the logical key.

Used to dedup embeddings and (image, spec, model) label calls so we never pay
for or recompute an identical query.
"""

from __future__ import annotations

import hashlib
import json
from pathlib import Path
from typing import Any

from veil_pgd.config import get_settings


class KVCache:
    def __init__(self, namespace: str, root: str | None = None):
        base = Path(root or get_settings().cache_dir) / namespace
        base.mkdir(parents=True, exist_ok=True)
        self._base = base

    def _path(self, key: str) -> Path:
        h = hashlib.sha256(key.encode("utf-8")).hexdigest()
        return self._base / f"{h}.json"

    def get(self, key: str) -> Any | None:
        p = self._path(key)
        if not p.exists():
            return None
        try:
            return json.loads(p.read_text())
        except json.JSONDecodeError:
            return None

    def set(self, key: str, value: Any) -> None:
        self._path(key).write_text(json.dumps(value))