File size: 2,691 Bytes
e4e4574 |
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 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 |
"""Caching layer for HuggingFace Crypto Data Engine"""
from __future__ import annotations
from typing import Optional, Any
from datetime import datetime, timedelta
import time
import json
from dataclasses import dataclass
@dataclass
class CacheEntry:
"""Cache entry with TTL"""
value: Any
expires_at: float
class MemoryCache:
"""In-memory cache with TTL support"""
def __init__(self):
self._cache: dict[str, CacheEntry] = {}
self._hits = 0
self._misses = 0
def get(self, key: str) -> Optional[Any]:
"""Get value from cache"""
if key not in self._cache:
self._misses += 1
return None
entry = self._cache[key]
# Check if expired
if time.time() > entry.expires_at:
del self._cache[key]
self._misses += 1
return None
self._hits += 1
return entry.value
def set(self, key: str, value: Any, ttl: int):
"""Set value in cache with TTL in seconds"""
expires_at = time.time() + ttl
self._cache[key] = CacheEntry(value=value, expires_at=expires_at)
def delete(self, key: str):
"""Delete key from cache"""
if key in self._cache:
del self._cache[key]
def clear(self):
"""Clear all cache entries"""
self._cache.clear()
self._hits = 0
self._misses = 0
def get_stats(self) -> dict:
"""Get cache statistics"""
total = self._hits + self._misses
hit_rate = (self._hits / total) if total > 0 else 0
return {
"size": len(self._cache),
"hits": self._hits,
"misses": self._misses,
"hitRate": round(hit_rate, 2)
}
def cleanup_expired(self):
"""Remove expired entries"""
current_time = time.time()
expired_keys = [
key for key, entry in self._cache.items()
if current_time > entry.expires_at
]
for key in expired_keys:
del self._cache[key]
# Global cache instance
cache = MemoryCache()
def cache_key(prefix: str, **kwargs) -> str:
"""Generate cache key from prefix and parameters"""
params = ":".join(f"{k}={v}" for k, v in sorted(kwargs.items()))
return f"{prefix}:{params}" if params else prefix
async def get_or_set(
key: str,
ttl: int,
factory: callable
) -> Any:
"""Get from cache or compute and store"""
# Try to get from cache
cached = cache.get(key)
if cached is not None:
return cached
# Compute value
value = await factory()
# Store in cache
cache.set(key, value, ttl)
return value
|