我有一个返回列表的函数,比如list_x。
def result(val):
..
return(list_x)
我每分钟调用result()并存储列表。
def other_func():
#called every minute
new_list = result(val)
我想将new_list的值存储一个小时(在某种内存缓存中可能是?)然后再次更新它,基本上是在一个小时后调用results(),而不是每分钟调用一次。我读到过functools.lru_cache,但我认为这在这里没有帮助。有什么想法吗?
发布于 2018-06-15 06:49:52
构建具有生存时间的单元素缓存非常简单:
_last_result_time = None
_last_result_value = None
def result(val):
global _last_result_time
global _last_result_value
now = datetime.datetime.now()
if not _last_result_time or now - _last_result_time > datetime.timedelta(hours=1):
_last_result_value = <expensive computation here>
_last_result_time = now
return _last_result_value
如果你想把它概括为一个装饰器,这并不难:
def cache(ttl=datetime.timedelta(hours=1)):
def wrap(func):
time, value = None, None
@functools.wraps(func)
def wrapped(*args, **kw):
nonlocal time
nonlocal value
now = datetime.datetime.now()
if not time or now - time > ttl:
value = func(*args, **kw)
time = now
return value
return wrapped
return wrap
如果你想让它处理不同的参数,为每个参数存储一个生存时间:
def cache(ttl=datetime.timedelta(hours=1)):
def wrap(func):
cache = {}
@functools.wraps(func)
def wrapped(*args, **kw):
now = datetime.datetime.now()
# see lru_cache for fancier alternatives
key = tuple(args), frozenset(kw.items())
if key not in cache or now - cache[key][0] > ttl:
value = func(*args, **kw)
cache[key] = (now, value)
return cache[key][1]
return wrapped
return wrap
当然,你可以给它增加一些特性--给它一个最大的大小,根据存储时间或者LRU或者其他任何你想要的东西来驱逐它,在装饰函数上公开缓存统计信息作为属性,等等。lru_cache
在stdlib中的实现应该会帮助你展示如何做大多数更棘手的事情(因为它几乎完成了所有的事情)。
发布于 2019-01-25 08:11:18
cachetools==3.1.0
中的ttl_cache
装饰器的工作方式与functools.lru_cache
很相似,但使用的是time to live。
import cachetools.func
@cachetools.func.ttl_cache(maxsize=128, ttl=10 * 60)
def example_function(key):
return get_expensively_computed_value(key)
class ExampleClass:
EXP = 2
@classmethod
@cachetools.func.ttl_cache()
def example_classmethod(cls, i):
return i* cls.EXP
@staticmethod
@cachetools.func.ttl_cache()
def example_staticmethod(i):
return i * 3
发布于 2018-06-15 06:54:08
装饰者通常能很好地解决这个问题。
def cache(fn=None,time_to_live=3600*24): # one DAY default (or whatever)
if not fn: return functools.partial(cache,time_to_live=time_to_live)
my_cache = {}
def _inner_fn(*args,**kwargs)
kws = sorted(kwargs.items()) # in python3.6+ you dont need sorted
key = tuple(args)+tuple(kw)
if key not in my_cache or time.time() > my_cache[key]['expires']:
my_cache[key] = {"value":fn(*args,**kwargs),"expires":time.time()+ time_to_live}
return my_cache[key]
return __inner_fn
@cache(time_to_live=3600) # an hour
def my_sqrt(x):
return x**0.5
@cache(time_to_live=60*30) # 30 mins
def get_new_emails():
return my_stmp.get_email_count()
顺便说一句,它内置于memcache中,这可能是一个更好的解决方案(我不确定您在哪个问题域中工作)
您还可以使用嵌套函数
def cache(time_to_live=3600*24): # one DAY default (or whatever)
def _wrap(fn):
my_cache = {}
def _inner_fn(*args,**kwargs)
kws = sorted(kwargs.items()) # in python3.6+ you dont need sorted
key = tuple(args)+tuple(kw)
if key not in my_cache or time.time() > my_cache[key]['expires']:
my_cache[key] = {"value":fn(*args,**kwargs),"expires":time.time()+ time_to_live}
return my_cache[key]
return _inner_fn
return _wrap
https://stackoverflow.com/questions/50866911
复制相似问题