|
|
|
|
|
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 |
|
|
|
|
|
|