-
Notifications
You must be signed in to change notification settings - Fork 1
/
cache_utils.py
63 lines (48 loc) · 1.79 KB
/
cache_utils.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
import threading
from datetime import timedelta
from functools import wraps
from typing import no_type_check
from django.utils.timezone import now
from posthog.settings import TEST
def cache_for(cache_time: timedelta, background_refresh=False):
def wrapper(fn):
@wraps(fn)
@no_type_check
def memoized_fn(*args, use_cache=not TEST, **kwargs):
if not use_cache:
return fn(*args, **kwargs)
current_time = now()
key = (args, frozenset(sorted(kwargs.items())))
def refresh():
try:
value = fn(*args, **kwargs)
memoized_fn._cache[key] = (now(), value)
memoized_fn._refreshing[key] = None
except Exception:
memoized_fn._refreshing[key] = None
raise
if key not in memoized_fn._cache:
refresh()
elif current_time - memoized_fn._cache[key][0] > cache_time:
if background_refresh:
if not memoized_fn._refreshing.get(key):
memoized_fn._refreshing[key] = current_time
t = threading.Thread(target=refresh)
t.start()
else:
refresh()
return memoized_fn._cache[key][1]
memoized_fn._cache = {}
memoized_fn._refreshing = {}
return memoized_fn
return wrapper
def instance_memoize(callback):
name = f"_{callback.__name__}_memo"
def _inner(self, *args):
if not hasattr(self, name):
setattr(self, name, {})
memo = getattr(self, name)
if args not in memo:
memo[args] = callback(self, *args)
return memo[args]
return _inner