"""In-memory translation session store with 24h TTL.""" import uuid from datetime import datetime, timezone, timedelta from typing import Any from backend.models import TranslationSession class SessionStore: """Thread-safe-ish in-memory dict for translation sessions. Each session is keyed by a UUID and has a 24-hour TTL. Expired sessions are lazily cleaned on access. """ def __init__(self, ttl_hours: int = 24): self._store: dict[str, dict[str, Any]] = {} self.ttl_hours = ttl_hours def create(self, session_data: TranslationSession) -> str: session_id = uuid.uuid4().hex[:16] self._store[session_id] = { "data": session_data.model_dump(), "created_at": datetime.now(timezone.utc), "updated_at": datetime.now(timezone.utc), } return session_id def get(self, session_id: str) -> TranslationSession | None: entry = self._store.get(session_id) if entry is None: return None # Check TTL now = datetime.now(timezone.utc) age = now - entry["updated_at"] if age > timedelta(hours=self.ttl_hours): del self._store[session_id] return None # Touch (refresh TTL) entry["updated_at"] = now return TranslationSession(**entry["data"]) def update(self, session_id: str, data: dict[str, Any]) -> TranslationSession | None: entry = self._store.get(session_id) if entry is None: return None entry["data"].update(data) entry["updated_at"] = datetime.now(timezone.utc) return TranslationSession(**entry["data"]) def delete(self, session_id: str) -> bool: if session_id in self._store: del self._store[session_id] return True return False def cleanup_expired(self): """Remove all expired sessions.""" now = datetime.now(timezone.utc) expired_keys = [ k for k, v in self._store.items() if (now - v["updated_at"]) > timedelta(hours=self.ttl_hours) ] for k in expired_keys: del self._store[k] # Singleton instance session_store = SessionStore(ttl_hours=24)