Andrew Yim
6bb8e6ab3c
Code review polish: - Type hints: List → List[Dict[str, Any]] for cache_get return and cache_put results parameter, since rerank produces dict-shaped entries. - TTL=0 short-circuits to None deterministically. Previously the test for ttl_seconds=0 passed only because clock motion between cache_put and cache_get made `time.time() - cached_at > 0` true. Now the semantics match the docstring intent. 8/8 tests still pass. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
74 lines
2.0 KiB
Python
74 lines
2.0 KiB
Python
"""Filesystem cache for rerank results, keyed by SHA256(query + candidate_ids)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
DEFAULT_CACHE_DIR = Path.home() / ".cache" / "notion-search"
|
|
DEFAULT_TTL_SECONDS = 86400 # 24 hours
|
|
|
|
|
|
def _cache_key(query: str, candidate_ids: List[str]) -> str:
|
|
payload = f"{query}|{','.join(sorted(candidate_ids))}"
|
|
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def cache_get(
|
|
query: str,
|
|
candidate_ids: List[str],
|
|
*,
|
|
cache_dir: Path = DEFAULT_CACHE_DIR,
|
|
ttl_seconds: int = DEFAULT_TTL_SECONDS,
|
|
) -> Optional[List[Dict[str, Any]]]:
|
|
"""Return cached results if file exists and is fresh, else None.
|
|
|
|
Corrupted cache files are silently treated as misses (and removed).
|
|
`ttl_seconds <= 0` always returns None (deterministic "always expired").
|
|
"""
|
|
if ttl_seconds <= 0:
|
|
return None
|
|
|
|
key = _cache_key(query, candidate_ids)
|
|
path = cache_dir / f"{key}.json"
|
|
if not path.exists():
|
|
return None
|
|
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (json.JSONDecodeError, OSError):
|
|
try:
|
|
path.unlink()
|
|
except OSError:
|
|
pass
|
|
return None
|
|
|
|
cached_at = data.get("cached_at_epoch", 0)
|
|
if time.time() - cached_at > ttl_seconds:
|
|
return None
|
|
|
|
return data.get("results")
|
|
|
|
|
|
def cache_put(
|
|
query: str,
|
|
candidate_ids: List[str],
|
|
results: List[Dict[str, Any]],
|
|
*,
|
|
cache_dir: Path = DEFAULT_CACHE_DIR,
|
|
) -> None:
|
|
"""Write results to cache. Creates cache_dir if missing."""
|
|
cache_dir.mkdir(parents=True, exist_ok=True)
|
|
key = _cache_key(query, candidate_ids)
|
|
path = cache_dir / f"{key}.json"
|
|
payload = {
|
|
"query": query,
|
|
"candidate_ids": list(candidate_ids),
|
|
"results": results,
|
|
"cached_at_epoch": time.time(),
|
|
}
|
|
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|