Source code for core_mixins.decorators.cache

# -*- coding: utf-8 -*-

"""Caching decorator for memoizing function results."""

import threading
from collections import OrderedDict
from functools import wraps
from typing import Any, Callable, Optional, Tuple


class _CacheWrapper:
    """Wrapper class to maintain cache for decorated function."""

    def __init__(self, fcn: Callable, maxsize: Optional[int] = None) -> None:
        self.fcn = fcn
        self.cache: OrderedDict[Tuple[Any, ...], Any] = OrderedDict()
        self._lock = threading.Lock()
        self.maxsize = maxsize
        wraps(fcn)(self)

    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        cache_key = args + tuple(sorted(kwargs.items()))
        with self._lock:
            if cache_key in self.cache:
                self.cache.move_to_end(cache_key)
                return self.cache[cache_key]

            self.cache[cache_key] = self.fcn(*args, **kwargs)

            if self.maxsize is not None and len(self.cache) > self.maxsize:
                self.cache.popitem(last=False)

            return self.cache[cache_key]




def cache(
    fcn: Optional[Callable] = None,
    *,
    maxsize: Optional[int] = None,
) -> Callable:
    """
    It maintains a cache of previous function call results that can be
    used to optimize the performance...

    :param fcn: The function being decorated.

    :param maxsize:
        Maximum number of entries to keep. When the limit is reached the
        least-recently-used entry is evicted. ``None`` (default) means
        the cache is unbounded.

    :return: The wrapped function.
    :rtype: Callable
    """

    def decorator(_fcn: Callable) -> _CacheWrapper:
        return _CacheWrapper(_fcn, maxsize=maxsize)

    if fcn is not None:
        return decorator(fcn)

    return decorator