使用@functools.lru_cache与字典参数

Question

我有一个方法，它需要一个字典作为参数之一。该方法解析字符串，并且字典提供了一些子字符串的替换值，因此它不需要是可变的。

由于这个函数会被频繁调用，而且针对冗余元素，所以我想到将其缓存起来可以提高效率。

但是，正如你可能已经猜到的那样，由于字典是可变的，因此无法使用@functools.lru_cache修饰我的函数。那么我该如何克服这个问题呢？

如果只使用标准库类和方法就能解决问题会更好。如果标准库中存在某种frozendict，我会非常开心（即使我没有看到过）。

PS：namedtuple只有在万不得已的情况下才能使用，因为它需要进行大量语法转换。

- Evpok

---

也许这可以帮助：https://dev59.com/OW445IYBdhLWcg3w1tui - mouad

我之前没有看到这个，但它并没有真正帮助到我。从头开始编写一个缓存装饰器在这里不值得花费精力，我想坚持使用标准库。无论如何，还是谢谢你 :) - Evpok

如何对namedtuple进行子类化并添加通过 x ["key"] 访问的功能？这可能只需要几行代码。 - Sven Marnach

我所知道的获取命名元组的唯一方法是通过调用工厂collections.namedtuple，它返回一个type。因此，如果我想要为命名元组添加__getitem__，我必须动态地进行操作，这应该是不可能的，即使它是可行的，也非常丑陋。还有其他的方法吗？ - Evpok

@Evpok：只需对namedtuple()返回的类型进行子类化：class X(namedtuple("Y", "a b c")): ...。 - Sven Marnach

@Sven Marnach 没想到这个。不错 :) - Evpok

Answer 1

使用这个而不是使用自定义的可哈希字典，避免重复造轮子！它是一个完全可哈希的冻结字典。

https://pypi.org/project/frozendict/

代码：

from frozendict import frozendict

def freezeargs(func):
    """Transform mutable dictionnary
    Into immutable
    Useful to be compatible with cache
    """

    @functools.wraps(func)
    def wrapped(*args, **kwargs):
        args = tuple([frozendict(arg) if isinstance(arg, dict) else arg for arg in args])
        kwargs = {k: frozendict(v) if isinstance(v, dict) else v for k, v in kwargs.items()}
        return func(*args, **kwargs)
    return wrapped

然后

@freezeargs
@lru_cache
def func(...):
    pass

代码来自于@fast_cen的答案

注意：此方法无法用于递归数据结构；例如，你可能有一个参数是列表，它是不可哈希的。你可以尝试将包装器设置为递归，这样它可以深入到数据结构中，并将每个dict冻结并将每个list转换为元组。

（我知道原始问题的提问者已经不需要解决方案了，但我曾经也在这里寻找同样的解决方案，所以留下这个信息给未来的人们）

- Cedar

---

@lru_cache 还是 lru_cache()？ - Itamar Mushkin

@lru_cache 是我一直在使用的。你所说的 lru_cache() 是什么意思？ - Cedar

Answer 2

这里有一个使用 @mhyfritz 技巧的装饰器。

def hash_dict(func):
    """Transform mutable dictionnary
    Into immutable
    Useful to be compatible with cache
    """
    class HDict(dict):
        def __hash__(self):
            return hash(frozenset(self.items()))

    @functools.wraps(func)
    def wrapped(*args, **kwargs):
        args = tuple([HDict(arg) if isinstance(arg, dict) else arg for arg in args])
        kwargs = {k: HDict(v) if isinstance(v, dict) else v for k, v in kwargs.items()}
        return func(*args, **kwargs)
    return wrapped

只需在您的lru_cache之前添加它即可。

@hash_dict
@functools.lru_cache()
def your_function():
    ...

- fast_cen

---

不允许在装饰的函数上调用clear_cache()或cache_info()。 - Nicholas Tulach

要提供clear_cache()和cache_info()函数的调用，只需在返回之前将这些函数添加到wrapped中。类似于wrapper.cache_info = func.cache_info和wrapper.cache_clear = func.cache_clear。 - rcsalvador

Answer 3

那么创建一个可散列的 dict 类如下：

class HDict(dict):
    def __hash__(self):
        return hash(frozenset(self.items()))

substs = HDict({'foo': 'bar', 'baz': 'quz'})
cache = {substs: True}

- mhyfritz

---

运作得非常好，虽然只有当字典的所有项都是可散列的时才有效，但这正是我的情况。不过，你有什么办法处理不可散列的 self.items() 吗？ - Evpok

目前我脑海中没有简单的方法。当然，你可以沿着字典递归下去，并在途中转换不可变对象（将字典转换为冻结集合，列表转换为元组等）... - mhyfritz

Answer 4

如何通过子类化namedtuple并添加x["key"]的访问方式？

class X(namedtuple("Y", "a b c")):
    def __getitem__(self, item):
        if isinstance(item, int):
            return super(X, self).__getitem__(item)
        return getattr(self, item)

- Sven Marnach

---

不错，但由于“keys”是用户定义的，这将强制我将其定义为调用方法的内部类，而我想避免这种情况。不过，这是个好主意，可能在某些时候很有用。 - Evpok

Answer 5

基于@Cedar的回答，按照建议为深度冻结添加递归：

def deep_freeze(thing):
    from collections.abc import Collection, Mapping, Hashable
    from frozendict import frozendict
    if thing is None or isinstance(thing, str):
        return thing
    elif isinstance(thing, Mapping):
        return frozendict({k: deep_freeze(v) for k, v in thing.items()})
    elif isinstance(thing, Collection):
        return tuple(deep_freeze(i) for i in thing)
    elif not isinstance(thing, Hashable):
        raise TypeError(f"unfreezable type: '{type(thing)}'")
    else:
        return thing


def deep_freeze_args(func):
    import functools

    @functools.wraps(func)
    def wrapped(*args, **kwargs):
        return func(*deep_freeze(args), **deep_freeze(kwargs))
    return wrapped

- Boris Lopez

Answer 6

在决定暂时放弃使用LRU缓存后，我们仍然想到了一个解决方案。这个装饰器使用JSON来序列化和反序列化发送到缓存的args/kwargs。可以与任意数量的args一起使用。将其作为函数的装饰器使用，而不是@lru_cache。最大大小设置为1024。

def hashable_lru(func):
    cache = lru_cache(maxsize=1024)

    def deserialise(value):
        try:
            return json.loads(value)
        except Exception:
            return value

    def func_with_serialized_params(*args, **kwargs):
        _args = tuple([deserialise(arg) for arg in args])
        _kwargs = {k: deserialise(v) for k, v in kwargs.items()}
        return func(*_args, **_kwargs)

    cached_function = cache(func_with_serialized_params)

    @wraps(func)
    def lru_decorator(*args, **kwargs):
        _args = tuple([json.dumps(arg, sort_keys=True) if type(arg) in (list, dict) else arg for arg in args])
        _kwargs = {k: json.dumps(v, sort_keys=True) if type(v) in (list, dict) else v for k, v in kwargs.items()}
        return cached_function(*_args, **_kwargs)
    lru_decorator.cache_info = cached_function.cache_info
    lru_decorator.cache_clear = cached_function.cache_clear
    return lru_decorator

- Harel

Answer 7

这是一个装饰器，可以像functools.lru_cache一样使用。但是它针对的是只接受一个参数的函数，该参数是一个具有可哈希值的扁平映射，并且具有固定的maxsize，最大为64。针对您的用例，您需要调整此示例或客户端代码。此外，要单独设置maxsize，必须实现另一个装饰器，但我还没有理解清楚，因为我不需要它。

from functools import (_CacheInfo, _lru_cache_wrapper, lru_cache,
                       partial, update_wrapper)
from typing import Any, Callable, Dict, Hashable

def lru_dict_arg_cache(func: Callable) -> Callable:
    def unpacking_func(func: Callable, arg: frozenset) -> Any:
        return func(dict(arg))

    _unpacking_func = partial(unpacking_func, func)
    _cached_unpacking_func = \
        _lru_cache_wrapper(_unpacking_func, 64, False, _CacheInfo)

    def packing_func(arg: Dict[Hashable, Hashable]) -> Any:
        return _cached_unpacking_func(frozenset(arg.items()))

    update_wrapper(packing_func, func)
    packing_func.cache_info = _cached_unpacking_func.cache_info
    return packing_func


@lru_dict_arg_cache
def uppercase_keys(arg: dict) -> dict:
    """ Yelling keys. """
    return {k.upper(): v for k, v in arg.items()}


assert uppercase_keys.__name__ == 'uppercase_keys'
assert uppercase_keys.__doc__ == ' Yelling keys. '
assert uppercase_keys({'ham': 'spam'}) == {'HAM': 'spam'}
assert uppercase_keys({'ham': 'spam'}) == {'HAM': 'spam'}
cache_info = uppercase_keys.cache_info()
assert cache_info.hits == 1
assert cache_info.misses == 1
assert cache_info.maxsize == 64
assert cache_info.currsize == 1
assert uppercase_keys({'foo': 'bar'}) == {'FOO': 'bar'}
assert uppercase_keys({'foo': 'baz'}) == {'FOO': 'baz'}
cache_info = uppercase_keys.cache_info()
assert cache_info.hits == 1
assert cache_info.misses == 3
assert cache_info.currsize == 3

如果想要更加通用的方法，可以使用第三方库中的装饰器@cachetools.cache，并将适当的函数设置为key。

- funky-future

Answer 8

解决方案可能更简单。lru_cache使用参数作为缓存的标识符，因此在字典的情况下，lru_cache不知道如何解释它。您可以将字典参数序列化为字符串，并在函数中反序列化回字典。效果很好。

该函数：

    @lru_cache(1024)
    def data_check(serialized_dictionary):
        my_dictionary = json.loads(serialized_dictionary)
        print(my_dictionary)

呼叫：

    data_check(json.dumps(initial_dictionary))

- Gregory R

Answer 9

在@Cedar answer的基础上进行扩展，添加递归冻结：

递归冻结：

def recursive_freeze(value):
    if isinstance(value, dict):
        for k,v in value.items():
            value[k] = recursive_freeze(v)
        return frozendict(value)
    else:
        return value

# To unfreeze
def recursive_unfreeze(value):
    if isinstance(value, frozendict):
        value = dict(value)
        for k,v in value.items():
            value[k] = recursive_unfreeze(v)
    
    return value

装饰器：

def freezeargs(func):
    """
    Transform mutable dictionnary into immutable.
    Useful to be compatible with cache
    """

    @functools.wraps(func)
    def wrapped(*args, **kwargs):
        args = tuple([recursive_freeze(arg) if isinstance(arg, dict) else arg for arg in args])
        kwargs = {k: recursive_freeze(v) if isinstance(v, dict) else v for k, v in kwargs.items()}
        return func(*args, **kwargs)
    return wrapped

- Kenneth Gabriel