Simple one size fits all Cache Wrapper compatible with FastApi endpoints

cache_wrapper.py

import logging
from inspect import signature
from typing import Callable, Any, Optional

import redis

cache = redis.Redis(host='localhost', port=6379, db=0)


class CacheReady:
    pass


class CacheWrapper(object):
    def __init__(
        self,
        func: Callable,
        timeout: int,
        arguments: bool = False,
    ):
        self._func = func
        self._timeout = timeout
        self._arguments = arguments

    @property
    def __annotations__(self):
        # This solves Typer injections
        return dict()

    @property
    def __signature__(self):
        # This solves endpoint parameters in swagger
        return signature(self._func)

    def get_key(self, *args, **kwargs):
        # Check if we should cache the response based on the arguments
        if self._arguments is True:

            # In order for objects to appear correctly and cache with inner parameters
            for key, value in kwargs.items():
                if value is not None and isinstance(value, CacheReady):
                    try:
                        kwargs[key] = value.__dict__
                    except AttributeError:
                        pass

            return f"{self._func.__name__}-{str(args)}-{str(kwargs)}"
        return f"{self._func.__name__}"

    def _call(self, *args, instance: Optional[Any] = None, **kwargs):
        _cache_key = self.get_key(*args, **kwargs)
        _result = cache.get(_cache_key)
        if _result.is_success:
            logging.info(
                "Cache hit",
                extra=dict(
                    function=self._func.__name__,
                    args=args,
                    kwargs=kwargs,
                ),
            )
            return _result.data

        if instance is None:
            _result = self._func(*args, **kwargs)
        else:
            _result = self._func(instance, *args, **kwargs)

        # Don't save exceptions in cache
        if not isinstance(_result, Exception):
            cache.set(_cache_key, _result, self._timeout)

        return _result

    @staticmethod
    def cache(timeout: int = 60, arguments: bool = False):

        # This wrapper adds support for passing the self parameter inside class functions
        def wrapper(func: Callable):
            _cache_wrapper = CacheWrapper(
                func=func, timeout=timeout, arguments=arguments
            )

            def inner_wrapper(instance: Optional[Any] = None, *args, **kwargs):
                return _cache_wrapper._call(*args, instance=instance, **kwargs)

            # Set inner function parameters to display properly in swagger
            inner_wrapper.__annotations__ = _cache_wrapper.__annotations__
            inner_wrapper.__signature__ = _cache_wrapper.__signature__
            inner_wrapper.__name__ = func.__name__
            inner_wrapper.__doc__ = func.__doc__

            return inner_wrapper

        return wrapper