A @cache_results decorator for functions that gives full control over bypassing/refreshing/clearing the cache

caching.py

import functools
import logging
from django.core.cache import DEFAULT_CACHE_ALIAS, caches

logger = logging.getLogger(__name__)


class CachedFunction:
    """
    The CachedFunction is a proxy object that implements cache-wrapper logic around
    an existing function. Use the :func:`cache_results` decorator to apply it.
    """

    def __init__(self, orig_function, key_function, alias=DEFAULT_CACHE_ALIAS):
        self.orig_function = orig_function
        self.key_function = key_function
        self.cache = caches[alias]

        # Same as @wraps(func) for classes
        functools.update_wrapper(self, orig_function)

    def __repr__(self):
        return f'<CachedFunction for {self.orig_function}>'

    def __call__(self, *args, **kwargs):
        """
        By calling the CachedFunction, it first checks for a cached value.
        When there isn't a cached value, the original function will be
        called and the cache is set.
        """
        # Fetch the data from the cache
        cache_key = self.cache_key(*args, **kwargs)
        value = self.cache.get(cache_key)

        # When data is missing, call the original function to retrieve it.
        if value is None or not self.is_expired(value):
            value = self.orig_function(*args, **kwargs)
            self.cache.set(cache_key, value)
        return value

    def cache_key(self, *args, **kwargs):
        """
        Generate the cache key. It receives all parameters
        that the original call would receive.
        This function can be overwritten.
        """
        return self.key_function(*args, **kwargs)

    def is_expired(self, value) -> bool:
        """
        Allow to invalidate cached results based on their returned data.
        For example, this allows to check for a timestamp or 'version' on the object.
        """
        return False

    def bypass_cache(self, *args, **kwargs):
        """
        Call the uncached function directly.
        """
        return self.orig_function(*args, **kwargs)

    def refresh_cache(self, *args, **kwargs):
        """
        Forcefully reset the cached data to the latest results of the function.
        """
        cache_key = self.cache_key(*args, **kwargs)
        value = self.orig_function(*args, **kwargs)
        self.cache.set(cache_key, value)
        logger.debug("Replacing cache value: %s", cache_key)
        return value

    def clear_cache(self, *args, **kwargs):
        """
        Clear the cached value of the function.
        """
        cache_key = self.cache_key(*args, **kwargs)
        logger.debug("Clearing cache key: %s", cache_key)
        self.cache.delete(cache_key)

    def from_cache(self, *args, **kwargs):
        """
        Retrieve the data from the cache only, returns None when missing.
        """
        cache_key = self.cache_key(*args, **kwargs)
        value = self.cache.get(cache_key, default=None)
        if value is None or self.is_expired(value):
            return None
        return value


def cache_results(key_function, wrapper=CachedFunction, alias=DEFAULT_CACHE_ALIAS):
    """
    Decorator that allows to cache a function,
    and also allows to skip the cache if needed.

    Usage::

        def key_function(arg1, arg2):
            return f"prefix.{arg1}.{arg2}"

        @cache_results(key_function=key_function)
        def some_function(arg1, arg2):
            return "COMPLEX DATA"

    Normal usage::

        value = some_function(1, 2)

    Skipping the cache::

        value = some_function.bypass_cache(1, 2)

    Taking only the cache value::

        value = some_function.from_cache(1, 2)

    Updating the cache::

        value = some_function.refresh_cache(1, 2)

    Fetch the key for manual action::

        cache_key = some_function.cache_key(1, 2)
    """
    def dec(func) -> CachedFunction:
        return wrapper(func, key_function=key_function, alias=alias)
    return dec