-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathcache.py
35 lines (30 loc) · 1018 Bytes
/
cache.py
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
from collections import OrderedDict
from threading import Lock
class Cache:
def __init__(self, capacity=100, expiration=300):
self.capacity = capacity
self.expiration = expiration
self.cache = OrderedDict()
self.lock = Lock()
def get(self, key):
with self.lock:
if key not in self.cache:
return None
item = self.cache[key]
if time.time() - item['timestamp'] > self.expiration:
del self.cache[key]
return None
self.cache.move_to_end(key)
return item['value']
def set(self, key, value):
with self.lock:
if key in self.cache:
del self.cache[key]
elif len(self.cache) >= self.capacity:
self.cache.popitem(last=False)
self.cache[key] = {'value': value, 'timestamp': time.time()}
def clear(self):
with self.lock:
self.cache.clear()
cache = Cache()