55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
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()
|