En utilisant @ functools.lru_cache avec des arguments dictionnaire

Question

J'ai une méthode qui prend (entre autres) un dictionnaire comme argument. La méthode est l'analyse des chaînes et le dictionnaire fournit un remplaçant pour certains sous-chaînes, il ne doit pas être mutable.

Cette fonction est appelée assez souvent, et sur des éléments redondants donc je me suis dit que la mise en cache, il améliorerait son efficacité.

Mais, comme vous avez pu le deviner, car dict est mutable et donc pas indexables, @functools.lru_cache ne peut pas décorer ma fonction. Alors, comment puis-je surmonter cela?

point de bonus si elle n'a besoin que de classes et méthodes de la bibliothèque standard. Idéalement, si elle existe une sorte de frozendict dans la bibliothèque standard que je ne l'ai pas vu ça ferait mon jour.

PS:. namedtuple seulement en dernier recours, car il aurait besoin d'un grand changement de syntaxe

Answer 1

Au lieu d'utiliser un dictionnaire hashable personnalisé, utilisez cela et éviter de réinventer la roue! Il est un dictionnaire congelé qui est tout hashable.

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

Code:

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

et

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

Code de prise de @fast_cen « s réponse

Note: cela ne fonctionne pas sur les structures de données récursives; par exemple, vous pourriez avoir un argument qui est une liste, qui est unhashable. Vous êtes invités à faire le récursive d'emballage, de telle sorte qu'il pénètre profondément dans la structure des données et fait chaque dict congelé et chaque tuple list.

(Je sais que nolonger OP veut une solution, mais je suis venu ici pour trouver la même solution, laissant ainsi ce pour les générations futures)

Answer 2

Qu'en est-il de créer une classe hashable de dict comme ceci:

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

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

Answer 3

Voici un décorateur ce truc de l'utilisation @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

Ajoutez simplement avant votre lru_cache.

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

Answer 4

Qu'en est-namedtuple et l'accès sous-classement add par 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)

Answer 5

Voici un décorateur qui peut être utilisé comme functools.lru_cache. Mais cela est à des fonctions qui pour cible ne prennent que un argument qui est un mapping plat avec valeurs HASHABLE et a une maxsize fixe 64. Pour votre cas d'utilisation que vous auriez à adapter cet exemple soit ou votre code client. En outre, pour régler la maxsize individuellement, il fallait mettre en œuvre un autre décorateur, mais je ne l'ai pas enveloppé ma tête depuis que je ne avais besoin.

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

Pour une approche plus générique pourrait utiliser le décorateur @ cachetools.cache d'un bibliothèque tierce partie avec un ensemble de fonctions appropriées en tant key.

Answer 6

Après avoir décidé d'abandonner le cache LRU pour notre cas d'utilisation pour l'instant, nous avons encore une solution. Celui-ci utilise décorateur JSON serialise et deserialise les args / kwargs envoyés dans le cache. Fonctionne avec un certain nombre de args. Utilisez-le comme décorateur sur une fonction au lieu de @lru_cache. la taille maximale est réglée à 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