cbpark84
1c919f1106
- run_batch(): sequential|concurrent|auto - chain(): 파이프라인 체이닝 - schema(): 스키마 자동 추출 - ResultCache: LRU+TTL 캐시 - preview_callback: 실시간 프리뷰 - upload_file/audio/video() - BridgeInputSEED: random/increment/fixed - MCPServer: MCP Tool 자동 생성
82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
"""ResultCache — LRU + TTL 인메모리 캐시."""
|
|
|
|
import time
|
|
from collections import OrderedDict
|
|
from typing import Any, Optional
|
|
|
|
|
|
class ResultCache:
|
|
"""
|
|
BridgeResult를 LRU + TTL로 캐시한다.
|
|
|
|
사용 예::
|
|
cache = ResultCache(max_size=100, ttl=3600)
|
|
cache.set("workflow_id_1", result)
|
|
result = cache.get("workflow_id_1") # None if expired
|
|
"""
|
|
|
|
def __init__(self, max_size: int = 100, ttl: float = 3600.0):
|
|
"""
|
|
Args:
|
|
max_size: 최대 캐시 항목 수 (초과 시 오래된 항목부터 제거)
|
|
ttl: 각 항목의 Time-To-Live (초), 0이면 만료 없음
|
|
"""
|
|
self.max_size = max_size
|
|
self.ttl = ttl
|
|
self._cache: OrderedDict[str, tuple[Any, float]] = OrderedDict()
|
|
|
|
def set(self, key: str, value: Any) -> None:
|
|
"""값을 캐시에 저장한다."""
|
|
if key in self._cache:
|
|
del self._cache[key]
|
|
|
|
self._cache[key] = (value, time.time())
|
|
|
|
if len(self._cache) > self.max_size:
|
|
self._cache.popitem(last=False)
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
"""캐시에서 값을 가져온다 (TTL 확인)."""
|
|
if key not in self._cache:
|
|
return default
|
|
|
|
value, stored_at = self._cache[key]
|
|
|
|
if self.ttl > 0 and time.time() - stored_at > self.ttl:
|
|
del self._cache[key]
|
|
return default
|
|
|
|
self._cache.move_to_end(key)
|
|
return value
|
|
|
|
def has(self, key: str) -> bool:
|
|
"""값이 캐시에 있고 유효한지 확인한다."""
|
|
if key not in self._cache:
|
|
return False
|
|
_, stored_at = self._cache[key]
|
|
if self.ttl > 0 and time.time() - stored_at > self.ttl:
|
|
del self._cache[key]
|
|
return False
|
|
return True
|
|
|
|
def delete(self, key: str) -> None:
|
|
"""캐시에서 값을 삭제한다."""
|
|
self._cache.pop(key, None)
|
|
|
|
def clear(self) -> None:
|
|
"""전체 캐시를 비운다."""
|
|
self._cache.clear()
|
|
|
|
def size(self) -> int:
|
|
"""유효한 항목의 개수를 반환한다."""
|
|
to_delete = []
|
|
for key, (_, stored_at) in self._cache.items():
|
|
if self.ttl > 0 and time.time() - stored_at > self.ttl:
|
|
to_delete.append(key)
|
|
for key in to_delete:
|
|
del self._cache[key]
|
|
return len(self._cache)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"ResultCache(size={self.size()}, max_size={self.max_size}, ttl={self.ttl}s)"
|