我有一个方法,它采用(除其他外)字典作为参数。该方法正在解析字符串,并且字典提供了某些子字符串的替换,因此它不必是可变的。
这个函数经常被调用,并且是在冗余元素上,所以我认为缓存它会提高它的效率。
但是,正如您可能已经猜到的那样,由于
dict
是可变的,因此不可散列,因此 @functools.lru_cache
无法装饰我的函数。那么我该如何克服这个问题呢?
如果只需要标准库类和方法,那就加分了。理想情况下,如果标准库中存在某种我没见过的
frozendict
,那会让我很开心。
PS:
namedtuple
仅在最后手段,因为它需要一个大的语法转变。
不要使用自定义的可哈希字典,而是使用它并避免重新发明轮子!这是一本冻结的字典,全部都是可哈希的。
https://pypi.org/project/frozendict/
代码:
from frozendict import frozendict
def freezeargs(func):
"""Convert a mutable dictionary 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
元组。
(我知道OP不再想要解决方案,但我来这里寻找相同的解决方案,所以把这个留给后代)
这是一个使用 @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():
...
创建一个可哈希的
dict
类怎么样:
class HDict(dict):
def __hash__(self):
return hash(frozenset(self.items()))
substs = HDict({'foo': 'bar', 'baz': 'quz'})
cache = {substs: True}
基于@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
如何子类化
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)
这是一个可以像
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
。
在决定暂时为我们的用例放弃 lru 缓存后,我们仍然想出了一个解决方案。该装饰器使用 json 来序列化和反序列化发送到缓存的 args/kwargs。适用于任意数量的参数。使用它作为函数的装饰器而不是@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
解决方案可能要简单得多。 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))
@Cedar 答案的扩展,添加递归冻结:
递归冻结:
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