Andrey K. Choi
c16ec54891
added: dictionary support for censore message/user management with dict triggers
43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
import time
|
||
from typing import Any, Optional, Dict
|
||
|
||
class TTLCache:
|
||
"""
|
||
Простой in-memory кеш с TTL и грубой политикой вытеснения (удаляем самый старый ключ).
|
||
Не потокобезопасен — вызывайте из одного потока/процесса.
|
||
"""
|
||
def __init__(self, ttl_seconds: int = 60, max_size: int = 1024):
|
||
self.ttl = ttl_seconds
|
||
self.max = max_size
|
||
self._data: Dict[Any, Any] = {}
|
||
self._ts: Dict[Any, float] = {}
|
||
|
||
def get(self, key: Any) -> Optional[Any]:
|
||
now = time.time()
|
||
ts = self._ts.get(key)
|
||
if ts is None:
|
||
return None
|
||
if now - ts > self.ttl:
|
||
# истёк TTL
|
||
self._data.pop(key, None)
|
||
self._ts.pop(key, None)
|
||
return None
|
||
return self._data.get(key)
|
||
|
||
def set(self, key: Any, value: Any):
|
||
# простое вытеснение — удаляем самый старый
|
||
if len(self._data) >= self.max and self._ts:
|
||
oldest_key = min(self._ts.items(), key=lambda kv: kv[1])[0]
|
||
self._data.pop(oldest_key, None)
|
||
self._ts.pop(oldest_key, None)
|
||
self._data[key] = value
|
||
self._ts[key] = time.time()
|
||
|
||
def invalidate(self, key: Any):
|
||
self._data.pop(key, None)
|
||
self._ts.pop(key, None)
|
||
|
||
def clear(self):
|
||
self._data.clear()
|
||
self._ts.clear()
|