import json import logging from blake3 import blake3 from redis import Redis from django.conf import settings _log = logging.getLogger(__name__) _log.addHandler(logging.NullHandler()) _redis = Redis() class Cache: """ Decorator that caches the result of a function It works by generating a key given the function signature and uses it to store the result of the function on redis, so if the function is called with the same parameters again it will be cached and will retrieve the data from redis and will not execute the function. It is assumed that the function will returns a dictionary that can be formatted to json. """ """If caching is enabled""" enabled = settings.CUSTOM_CACHE['enabled'] """Time to expire, by default a week""" expire = 60 * 60 * 24 * 7 def __init__(self, function): self.function = function def __call__(self, *args, **kwargs): if not self.enabled: _log.info('Cache is disabled, executing function directly') return self.function(*args, **kwargs) _log.info(f'Caching function {self.function.__name__} with argument list {args} and dictionary {kwargs}') key = f'{self.function.__name__}:{self.create_hash_key(args, kwargs)}' _log.debug(f'Resolved key for function is "{key}"') if _redis.exists(key): _log.info(f'Key was in cache') result = json.loads(_redis.get(key)) return result else: _log.info('Key was not in cache') result = self.function(*args, **kwargs) _redis.set(key, json.dumps(result), ex=self.expire) return result @staticmethod def create_hash_key(args, kwargs): return blake3((str(args) + str(kwargs)).encode('utf-8')).hexdigest()