早些时候,我也遇到了这个问题。最后,我使用了 Redis。
并且在 中 werkeug
,有一个缓存库,这使得 Redis 易于使用。
from werkzeug.contrib.cache import RedisCache
有关更多信息,请参阅文档
顺便说一句,如果您的应用程序在单进程中运行(多线程也可以),您可以使用下面的代码。
class CachedItem:
def __init__(self, item, duration):
self.item = item
self.duration = duration
self.time_stamp = time.time()
def __repr__(self):
return '<CachedItem {%s} expires at: %s>' % (self.item, time.time() + self.duration)
class CachedDict(dict):
def __init__(self, *args, **kwargs):
super(CachedDict, self).__init__(*args, **kwargs)
self.lock = threading.Lock()
def get_cache(self, key, default=None, duration=300):
with self.lock:
self._clean()
if key in self:
return self[key].item
else:
self[key] = CachedItem(default, duration)
return self[key].item
def set_cache(self, key, value, duration=300):
with self.lock:
self[key] = CachedItem(value, duration)
def _clean(self):
for key in list(self.keys()): # [self.keys()] error, we get dict_keys type
if self[key].time_stamp + self[key].duration <= time.time():
self.pop(key)