可以替换或删除 functools.lru_cache 中的特定密钥吗?

Can One Replace or Remove a specific key from functools.lru_cache?

我正在使用 functools.lru_cache 在给定特定输入 * 的情况下提供临时文件路径。但是,如果路径不再存在,我想 remove/replace 单个对应的键。 cache_clear() 方法有点矫枉过正,cache_info() 似乎没有帮助。

感谢您的帮助!

* 正在缓存的方法将文件对象从 S3 流式传输到本地临时文件。

这是通过 python.org 问题 allow to cache_clear(some_key) in lru_cache 请求的,但被拒绝了。因此 lru_cache 中没有清除特定条目的方法。

从该问题链接有一个很好的建议,使用名为 Foundation for rolling your own LRU cache variantscollections.OrderedDict 实现您自己的变体。

使用ring.lru通过键控制缓存数据

import ring


@ring.lru()
def f(path):
    print('fetching', path)
    return 'some contents in ' + path


if __name__ == '__main__':
    f('path1')
    f('path1')  # no print
    f.delete('path1')
    f('path1')  # print again

还有setupdate可以替换

这是我的解决方案,根据@dmulter 的建议改编自 Foundation for rolling your own LRU cache variants

我添加了 update_wrapper 并覆盖了 __repr__ 以使其作为 func 的包装器更好地工作。然后添加了一些方法来从缓存中删除项目(通过传递调用它的相同参数),或直接替换缓存中的值(通过传递要替换的值,然后传递您用于替换的相同参数)再次调用函数)。我还通过对参数进行散列来添加对 **kwargs 的一些支持。我还没有非常彻底地测试它,但它似乎工作得很好。希望这对下一个寻找此功能的人有用。

我觉得这个应该可以像@lru_cache一样使用。

from functools import update_wrapper
    class MyLRU:
        def __init__(self, func, maxsize=128):
            self.cache = collections.OrderedDict()
            self.func = func
            self.maxsize = maxsize
            update_wrapper(self, self.func)
    
        def __call__(self, *args, **kwargs):
            cache = self.cache
            key = self._generate_hash_key(*args, **kwargs)
            if key in cache:
                cache.move_to_end(key)
                return cache[key]
            result = self.func(*args, **kwargs)
            cache[key] = result
            if len(cache) > self.maxsize:
                cache.popitem(last=False)
            return result
    
        def __repr__(self):
            return self.func.__repr__()
    
        def clear_cache(self):
            self.cache.clear()
    
        def cache_remove(self, *args, **kwargs):
            """Remove an item from the cache by passing the same args and kwargs"""
            key = self._generate_hash_key(*args, **kwargs)
            if key in self.cache:
                self.cache.pop(key)
    
        def cache_replace(self, value, *args, **kwargs):
            key = self._generate_hash_key(*args, **kwargs)
            self.cache[key] = value
    
        @staticmethod
        def _generate_hash_key(*args, **kwargs):
            key = hash(args)+hash(frozenset(sorted(kwargs.items())))
            return key

例如

@MyLRU
def test(a, b=1, c=2):
    print(f'a={a}, b={b}, c={c}')
    return (a,b,c)

test(1, 2, 3)  # prints and returns
test(1, 2, 3)  # only returns (i.e. using cache)
test(1, b=2, c=3)  # still prints, but this is a limitation with lru_cache as well
test(1, c=3, b=2)  # only returns (an improvement on lru_cache behaviour I think)
test(3, 4, 5)  # prints and returns
test.cache_remove(1,2,3)
test(1, 2, 3)  # prints again
test(3, 4, 5)  # only returns (so the rest of the cache wasn't cleared)
test.cache_replace('hi there', 1, 2, 3)
test(1, 2, 3)  # only returns 'hi there' (so cache was replaced)