-
Notifications
You must be signed in to change notification settings - Fork 0
/
cache.py
64 lines (53 loc) · 1.74 KB
/
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
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
"""
In-memory caching of key-value pairs with extra keys.
"""
class Cache:
"""
Implements a flexible generic storage mechanism with multiple indexes to access data.
"""
def __init__(self, name):
self.name = name
self.data = {}
self.index = {}
def clear(self):
self.index = {}
self.data = {}
def add(self, pk, value, extrakeys=None):
try:
self.data[pk] = value
if extrakeys is not None:
for k in extrakeys.keys():
if k not in self.index:
self.index[k] = {}
v = extrakeys[k]
if v not in self.index[k]:
self.index[k][v] = [pk]
else:
self.index[k][v].append(pk)
except IndexError as e:
print(f"Cache {self.name} add error {e} pk:{pk} extrakeys: {extrakeys} value {value}")
def get(self, pk):
return self.data[pk]
def get_with_default(self, pk, default=None):
if pk not in self.data:
return default
return self.data[pk]
def get_by_unique_key(self, key, value):
if key not in self.index:
return None
if value not in self.index[key]:
return None
key_list = self.index[key][value]
if len(key_list) != 1:
return None
return self.data[key_list[0]]
def list_by_key(self, key, value):
if key not in self.index:
return None
if value not in self.index[key]:
return None
keylist = self.index[key][value]
valuelist = []
for pk in keylist:
valuelist.append(self.data[pk])
return valuelist