| |
| import time |
| from typing import Optional |
|
|
| class SessionCache: |
| def __init__(self): |
| self.memory_cache = {} |
| self.hits = 0 |
| self.misses = 0 |
| |
| def get(self, session_id: str) -> Optional[dict]: |
| if session_id in self.memory_cache: |
| self.hits += 1 |
| return self.memory_cache[session_id] |
| self.misses += 1 |
| return None |
| |
| def set(self, session_id: str, data: dict, ttl: int = 3600): |
| |
| if self._get_total_size() > 100 * 1024 * 1024: |
| self._evict_oldest() |
| |
| compressed_data = self._compress_data(data) |
| self.memory_cache[session_id] = { |
| 'data': compressed_data, |
| 'timestamp': time.time(), |
| 'ttl': ttl |
| } |
| |
| def delete(self, session_id: str): |
| """ |
| Remove session from cache |
| """ |
| if session_id in self.memory_cache: |
| del self.memory_cache[session_id] |
| |
| def clear(self): |
| """ |
| Clear all cached sessions |
| """ |
| self.memory_cache.clear() |
| self.hits = 0 |
| self.misses = 0 |
| |
| def get_hit_rate(self) -> float: |
| """ |
| Calculate cache hit rate |
| """ |
| total = self.hits + self.misses |
| return self.hits / total if total > 0 else 0.0 |
| |
| def _get_total_size(self) -> int: |
| """ |
| Calculate total size of cached data |
| """ |
| |
| return len(str(self.memory_cache)) |
| |
| def _evict_oldest(self): |
| """ |
| Evict oldest session based on timestamp |
| """ |
| if not self.memory_cache: |
| return |
| |
| oldest_session = min( |
| self.memory_cache.items(), |
| key=lambda x: x[1].get('timestamp', 0) |
| ) |
| del self.memory_cache[oldest_session[0]] |
| |
| def _compress_data(self, data: dict) -> dict: |
| """ |
| Compress data using specified compression algorithm |
| """ |
| |
| |
| return data |
|
|
|
|
