forked from bwasti/cache.py
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcache.py
More file actions
134 lines (104 loc) · 2.83 KB
/
cache.py
File metadata and controls
134 lines (104 loc) · 2.83 KB
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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import pickle
import time
import inspect
import base64
import hashlib
debug = False
def log(s):
if debug:
print(s)
caches = dict()
updated_caches = []
def get_cache(fname):
if fname in caches:
return caches[fname]
try:
with open(fname, "rb") as f:
c = pickle.load(f)
except:
c = dict()
caches[fname] = c
return c
def write_to_cache(fname, obj):
updated_caches.append(fname)
caches[fname] = obj
def cleanup():
for fname in updated_caches:
with open(fname, "wb") as f:
pickle.dump(caches[fname], f)
def get_fn_hash(f):
return base64.b64encode(hashlib.sha1(inspect.getsource(f).encode("utf-8")).digest())
NONE = 0
ARGS = 1
KWARGS = 2
def cache(fname=".cache.pkl", timeout=-1, key=ARGS | KWARGS):
def impl(fn):
load_t = time.time()
c = get_cache(fname)
log("loaded cache in {:.2f}s".format(time.time() - load_t))
def d(*args, **kwargs):
log("checking cache on {}".format(fn.__name__))
if key == ARGS | KWARGS:
k = pickle.dumps((fn.__name__, args, kwargs))
if key == ARGS:
k = pickle.dumps((fn.__name__, args))
if key == KWARGS:
k = pickle.dumps((fn.__name__, kwargs))
if key == NONE:
k = pickle.dumps((fn.__name__))
if k in c:
h, t, to, res = c[k]
if get_fn_hash(fn) == h and (to < 0 or (time.time() - t) < to):
log("cache hit.")
return res
log("cache miss.")
res = fn(*args, **kwargs)
c[k] = (get_fn_hash(fn), time.time(), timeout, res)
save_t = time.time()
write_to_cache(fname, c)
log("saved cache in {:.2f}s".format(time.time() - save_t))
return res
return d
return impl
@cache(timeout=0.2)
def expensive(k):
time.sleep(0.2)
return k
@cache(key=KWARGS)
def expensive2(k, kwarg1=None):
time.sleep(0.2)
return k
def test():
# Test timeout
t = time.time()
v = expensive(1)
assert v == 1
assert time.time() - t > 0.1
t = time.time()
expensive(1)
assert time.time() - t < 0.1
time.sleep(0.3)
t = time.time()
expensive(1)
assert time.time() - t > 0.1
t = time.time()
v = expensive(2)
assert v == 2
assert time.time() - t > 0.1
# Test key=_ annotation
t = time.time()
v = expensive2(2, kwarg1="test")
assert v == 2
assert time.time() - t > 0.1
t = time.time()
v = expensive2(1, kwarg1="test")
assert v == 2
assert time.time() - t < 0.1
t = time.time()
v = expensive2(1, kwarg1="test2")
assert v == 1
assert time.time() - t > 0.1
cleanup()
print("pass")
if __name__ == "__main__":
test()