src/cpptai/io/cache.py

from __future__ import annotations

import hashlib
import json
from dataclasses import asdict, is_dataclass
from pathlib import Path
from typing import Any, Dict, Optional


def _normalize_payload(payload: Any) -> str:
    if is_dataclass(payload):
        payload = asdict(payload)
    try:
        return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    except (TypeError, ValueError):
        return str(payload)


def _key_to_path(cache_dir: Path, namespace: str, key: str) -> Path:
    h = hashlib.sha256(key.encode("utf-8")).hexdigest()
    return cache_dir / f"{namespace}_{h}.json"


def cache_get(cache_dir: str, namespace: str, key_payload: Any) -> Optional[Dict[str, Any]]:
    cdir = Path(cache_dir)
    cdir.mkdir(parents=True, exist_ok=True)
    key = _normalize_payload(key_payload)
    path = _key_to_path(cdir, namespace, key)
    if not path.exists():
        return None
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, FileNotFoundError, IOError):
        return None


def cache_set(cache_dir: str, namespace: str, key_payload: Any, value: Dict[str, Any]) -> None:
    cdir = Path(cache_dir)
    cdir.mkdir(parents=True, exist_ok=True)
    key = _normalize_payload(key_payload)
    path = _key_to_path(cdir, namespace, key)
    try:
        path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
    except (IOError, OSError):
        return