File size: 975 Bytes
8ff9278 |
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 |
import time
import asyncio
from typing import Any, Dict, Optional, Tuple
class CacheManager:
def __init__(self):
self._cache: Dict[str, Tuple[Any, float]] = {}
self._lock = asyncio.Lock()
async def get(self, key: str) -> Optional[Any]:
async with self._lock:
if key in self._cache:
value, expiry = self._cache[key]
if time.time() < expiry:
return value
else:
del self._cache[key]
return None
async def set(self, key: str, value: Any, ttl: int = 60):
async with self._lock:
self._cache[key] = (value, time.time() + ttl)
async def delete(self, key: str):
async with self._lock:
if key in self._cache:
del self._cache[key]
async def clear(self):
async with self._lock:
self._cache.clear()
# Global cache instance
cache_manager = CacheManager()
|