Source code for lalandre_rag.retrieval.result_cache

"""
Redis-backed result cache for the retrieval service.
"""

import dataclasses
import hashlib
import json
import logging
from typing import Any, Dict, List, Optional

from .result import RetrievalResult

logger = logging.getLogger(__name__)




class RetrievalCache:
    """Thin wrapper around Redis for caching retrieval results."""

    def __init__(self, redis_client: Any, ttl: int) -> None:
        self._redis = redis_client
        self._ttl = ttl

    @property
    def enabled(self) -> bool:
        """Return whether Redis caching is currently active."""
        return self._redis is not None and self._ttl > 0



    @staticmethod
    def cache_key(
        query: str,
        top_k: int,
        score_threshold: Optional[float],
        filters: Optional[Dict[str, Any]],
        granularity: Optional[str],
        collections: Optional[List[str]],
        embedding_preset: Optional[str] = None,
    ) -> str:
        """Build a stable cache key for one retrieval request shape."""
        payload = json.dumps(
            {
                "q": query,
                "k": top_k,
                "thr": score_threshold,
                "f": sorted(filters.items()) if filters else [],
                "g": granularity,
                "c": sorted(collections or []),
                "ep": embedding_preset or "",
            },
            sort_keys=True,
        )
        return f"retrieval:{hashlib.sha256(payload.encode()).hexdigest()[:24]}"




    def get(self, key: str) -> Optional[List[RetrievalResult]]:
        """Fetch cached retrieval results for *key*, if present."""
        if not self.enabled:
            return None
        try:
            raw = self._redis.get(key)
            if not raw:
                return None
            return [RetrievalResult(**item) for item in json.loads(raw)]
        except Exception as exc:
            logger.warning("Retrieval cache GET failed for %s: %s", key[:40], exc)
            return None




    def set(self, key: str, results: List[RetrievalResult]) -> None:
        """Store retrieval results for *key* with the configured TTL."""
        if not self.enabled or not results:
            return
        try:
            payload = json.dumps([dataclasses.asdict(r) for r in results])
            self._redis.setex(key, self._ttl, payload)
        except Exception as exc:
            logger.warning("Retrieval cache SET failed for %s: %s", key[:40], exc)