在python中使用时间限制在内存中缓存

Question

我有一个返回列表的函数,比如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但我认为这在这里无济于事.有任何想法吗？

Answer 1

将ttl_cache在装饰cachetools==3.1.0作品很像functools.lru_cache，但有生存时间。

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

Answer 2

构建具有生存时间的单元素缓存非常简单：

_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_cachestdlib 中的实现应该有所帮助向您展示如何做大多数棘手的事情（因为它几乎可以完成所有这些事情）。

Answer 3

装饰师通常很好地解决了这个问题

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