55 lines
1.4 KiB
Python
55 lines
1.4 KiB
Python
|
# retoor <retoor@molodetz.nl>
|
||
|
|
import threading
|
||
|
|
import time
|
||
|
|
|
||
|
|
|
||
|
|
class TTLCache:
|
||
|
|
def __init__(self, ttl=60, max_size=1000):
|
||
|
|
self.ttl = ttl
|
||
|
|
self.max_size = max_size
|
||
|
|
self._data = {}
|
||
|
|
self._lock = threading.Lock()
|
||
|
|
|
||
|
|
def get(self, key, default=None):
|
||
|
|
if self.ttl <= 0:
|
||
|
|
return default
|
||
|
|
with self._lock:
|
||
|
|
item = self._data.get(key)
|
||
|
|
if item is None:
|
||
|
|
return default
|
||
|
|
expires, value = item
|
||
|
|
if expires < time.monotonic():
|
||
|
|
self._data.pop(key, None)
|
||
|
|
return default
|
||
|
|
return value
|
||
|
|
|
||
|
|
def set(self, key, value):
|
||
|
|
if self.ttl <= 0:
|
||
|
|
return
|
||
|
|
with self._lock:
|
||
|
|
if len(self._data) >= self.max_size:
|
||
|
|
self._evict()
|
||
|
|
self._data[key] = (time.monotonic() + self.ttl, value)
|
||
|
|
|
||
|
|
def pop(self, key):
|
||
|
|
with self._lock:
|
||
|
|
self._data.pop(key, None)
|
||
|
|
|
||
|
|
def clear(self):
|
||
|
|
with self._lock:
|
||
|
|
self._data.clear()
|
||
|
|
|
||
|
|
def __contains__(self, key):
|
||
|
|
return self.get(key, _MISSING) is not _MISSING
|
||
|
|
|
||
|
|
def _evict(self):
|
||
|
|
now = time.monotonic()
|
||
|
|
expired = [k for k, (exp, _) in self._data.items() if exp < now]
|
||
|
|
for key in expired:
|
||
|
|
self._data.pop(key, None)
|
||
|
|
while len(self._data) >= self.max_size:
|
||
|
|
self._data.pop(next(iter(self._data)))
|
||
|
|
|
||
|
|
|
||
|
|
_MISSING = object()
|