Spaces:
Running
Running
File size: 2,181 Bytes
5cceba0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | """Short-TTL in-memory cache for /analyze — identical cases return identical JSON."""
from __future__ import annotations
import hashlib
import json
import threading
import time
from typing import Any
from .config import settings
from .schemas import AnalyzeRequest, CaseAnalysisResponse
_lock = threading.Lock()
_store: dict[str, tuple[float, dict[str, Any]]] = {}
def case_fingerprint(req: AnalyzeRequest) -> str:
"""Stable hash of the clinician-visible analyze inputs."""
payload = {
"clinical_summary": " ".join((req.clinical_summary or "").split()),
"encounter_type": (req.encounter_type or "").strip(),
"province_code": (req.province_code or "").strip().upper(),
"provider_specialty_code": (req.provider_specialty_code or "").strip(),
"patient_age_months": req.patient_age_months,
"time_of_service": (req.time_of_service or "").strip(),
"time_spent_minutes": req.time_spent_minutes,
"service_date": (req.service_date or "").strip(),
"is_holiday": bool(req.is_holiday),
}
raw = json.dumps(payload, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
def cache_get(key: str) -> CaseAnalysisResponse | None:
ttl = settings.analyze_cache_ttl_seconds
if ttl <= 0:
return None
now = time.monotonic()
with _lock:
item = _store.get(key)
if not item:
return None
expires_at, payload = item
if expires_at <= now:
_store.pop(key, None)
return None
return CaseAnalysisResponse.model_validate(payload)
def cache_put(key: str, result: CaseAnalysisResponse) -> None:
ttl = settings.analyze_cache_ttl_seconds
if ttl <= 0:
return
expires_at = time.monotonic() + float(ttl)
payload = result.model_dump(mode="json")
with _lock:
_store[key] = (expires_at, payload)
# Soft bound — drop oldest ~half when oversized.
if len(_store) > 512:
ordered = sorted(_store.items(), key=lambda kv: kv[1][0])
for stale_key, _ in ordered[:256]:
_store.pop(stale_key, None)
|