-
Notifications
You must be signed in to change notification settings - Fork 1
/
local_cache.py
78 lines (59 loc) · 1.92 KB
/
local_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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
# Copyright 2024 UW-IT, University of Washington
# SPDX-License-Identifier: Apache-2.0
from threading import currentThread
LOCAL_CACHE = {}
class local_cache(object):
"""
A decorator or context manager that will cache some values for the
lifetime of the decorator/context manager.
"""
def _function(self, func, *args, **kwargs):
def wrapped(*args, **kwargs):
try:
self.__enter__()
value = func(*args, **kwargs)
finally:
self.__exit__()
return value
return wrapped
# This handles function decorators + class decorators. Needs to switch
# on argument type :(
def __call__(self, decorated, *args, **kwargs):
if isinstance(decorated, type):
raise Exception("Can only decorate a function, not a class")
return self._function(decorated, *args, **kwargs)
# These handle context managers, and are used by the others
def __enter__(self, *args, **kwargs):
thread = currentThread()
LOCAL_CACHE[thread] = {}
return self
def __exit__(*args, **kwargs):
thread = currentThread()
del LOCAL_CACHE[thread]
def _get_local_cache():
thread = currentThread()
if thread in LOCAL_CACHE:
return LOCAL_CACHE[thread]
if hasattr(thread, 'parent'):
thread = thread.parent
if thread in LOCAL_CACHE:
return LOCAL_CACHE[thread]
return
def set_cache_value(key, value):
"""
If local cache is being used, this will set a cache value for the
lifetime of that cache.
"""
cache = _get_local_cache()
if cache is None:
return
cache[key] = value
def get_cache_value(key):
"""
If local cache is being used, this will get the cache value, if it exists.
"""
cache = _get_local_cache()
if cache is None:
return
if key in cache:
return cache[key]