"""캐시 관리 - 메모리 기반 캐시 구현"""
import hashlib
import json
import time
from typing import Any, TypeVar
from .types import CacheStats
T = TypeVar("T")
class CacheEntry:
"""캐시 엔트리"""
def __init__(self, value: Any, ttl: int):
self.value = value
self.expires_at = time.time() + ttl
def is_expired(self) -> bool:
"""만료 여부 확인"""
return time.time() > self.expires_at
class CacheManager:
"""메모리 기반 캐시 매니저"""
def __init__(self, cleanup_interval: int = 300):
"""
Args:
cleanup_interval: 만료된 캐시 정리 주기 (초)
"""
self._cache: dict[str, CacheEntry] = {}
self._hits = 0
self._misses = 0
self._cleanup_interval = cleanup_interval
self._last_cleanup = time.time()
def _maybe_cleanup(self) -> None:
"""필요시 만료된 캐시 정리"""
now = time.time()
if now - self._last_cleanup > self._cleanup_interval:
self._cleanup()
self._last_cleanup = now
def _cleanup(self) -> None:
"""만료된 캐시 엔트리 삭제"""
expired_keys = [
key for key, entry in self._cache.items()
if entry.is_expired()
]
for key in expired_keys:
del self._cache[key]
def get(self, key: str) -> Any | None:
"""
캐시에서 값 조회
Args:
key: 캐시 키
Returns:
캐시된 값 또는 None
"""
self._maybe_cleanup()
entry = self._cache.get(key)
if entry is None:
self._misses += 1
return None
if entry.is_expired():
del self._cache[key]
self._misses += 1
return None
self._hits += 1
return entry.value
def set(self, key: str, value: Any, ttl: int) -> None:
"""
캐시에 값 저장
Args:
key: 캐시 키
value: 저장할 값
ttl: TTL (초)
"""
self._cache[key] = CacheEntry(value, ttl)
def delete(self, key: str) -> bool:
"""
캐시에서 값 삭제
Args:
key: 캐시 키
Returns:
삭제 여부
"""
if key in self._cache:
del self._cache[key]
return True
return False
def clear(self) -> None:
"""캐시 전체 삭제"""
self._cache.clear()
self._hits = 0
self._misses = 0
def get_stats(self) -> CacheStats:
"""캐시 통계 조회"""
return CacheStats(
hits=self._hits,
misses=self._misses,
size=len(self._cache),
)
def generate_cache_key(endpoint: str, params: dict[str, Any] | None = None) -> str:
"""
캐시 키 생성
Args:
endpoint: API 엔드포인트
params: 쿼리 파라미터
Returns:
캐시 키 (해시)
"""
# 파라미터 정렬하여 일관된 키 생성
sorted_params = json.dumps(params or {}, sort_keys=True, ensure_ascii=False)
key_string = f"{endpoint}:{sorted_params}"
# MD5 해시로 키 생성 (충돌 가능성 낮고 빠름)
return hashlib.md5(key_string.encode()).hexdigest()
