Module licenseware.decorators.caching
Expand source code
import pickle
from functools import wraps
from typing import Callable
from licenseware.dependencies.redis_cache import RedisCache
from licenseware.utils.common import get_http_request_tenant_id
caching_database = RedisCache()
TEN_MINUTES = 600
def _serialize(obj):
return pickle.dumps(obj)
def _deserialize(obj):
return pickle.loads(obj)
def _hash_args(*args, **kwargs):
return _serialize((args, kwargs))
def _lookup_value(key):
value = caching_database.get(key)
if value:
return _deserialize(value)
def _save_result(key, result, expiry):
return caching_database.set(key, _serialize(result), expiry)
def cache_result(fn: Callable = None, expiry: int = TEN_MINUTES) -> Callable:
"""
Decorator to cache the result of a flask request.
"""
def decorator(fn: Callable) -> Callable:
@wraps(fn)
def wrapper(*args, **kwargs) -> Callable:
hashed_request = _hash_args(*args, **kwargs)
cached_result = _lookup_value(hashed_request)
if cached_result:
return cached_result
result = fn(*args, **kwargs)
_save_result(hashed_request, result, expiry)
_save_key_to_tenant_id(hashed_request)
return result
return wrapper
return decorator(fn) if fn else decorator
def _save_key_to_tenant_id(key: bytes, tenant_id: str = None):
if tenant_id is None:
tenant_id = get_http_request_tenant_id()
if tenant_id is not None:
caching_database.sadd(tenant_id, key)
def clear_caches_for_tenant_id(tenant_id: str = None):
if tenant_id is None:
tenant_id = get_http_request_tenant_id()
if tenant_id is not None:
keys = caching_database.smembers(tenant_id)
caching_database.delete(tenant_id, *keys)
Functions
def cache_result(fn: Callable = None, expiry: int = 600) ‑> Callable
-
Decorator to cache the result of a flask request.
Expand source code
def cache_result(fn: Callable = None, expiry: int = TEN_MINUTES) -> Callable: """ Decorator to cache the result of a flask request. """ def decorator(fn: Callable) -> Callable: @wraps(fn) def wrapper(*args, **kwargs) -> Callable: hashed_request = _hash_args(*args, **kwargs) cached_result = _lookup_value(hashed_request) if cached_result: return cached_result result = fn(*args, **kwargs) _save_result(hashed_request, result, expiry) _save_key_to_tenant_id(hashed_request) return result return wrapper return decorator(fn) if fn else decorator
def clear_caches_for_tenant_id(tenant_id: str = None)
-
Expand source code
def clear_caches_for_tenant_id(tenant_id: str = None): if tenant_id is None: tenant_id = get_http_request_tenant_id() if tenant_id is not None: keys = caching_database.smembers(tenant_id) caching_database.delete(tenant_id, *keys)