Spaces:
Running
Running
| """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) | |