84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
# Python imports
|
|
from functools import wraps
|
|
|
|
# Django imports
|
|
from django.conf import settings
|
|
from django.core.cache import cache
|
|
|
|
# Third party imports
|
|
from rest_framework.response import Response
|
|
|
|
|
|
def generate_cache_key(custom_path, auth_header=None):
|
|
"""Generate a cache key with the given params"""
|
|
if auth_header:
|
|
key_data = f"{custom_path}:{auth_header}"
|
|
else:
|
|
key_data = custom_path
|
|
return key_data
|
|
|
|
|
|
def cache_response(timeout=60 * 60, path=None, user=True):
|
|
"""decorator to create cache per user"""
|
|
|
|
def decorator(view_func):
|
|
@wraps(view_func)
|
|
def _wrapped_view(instance, request, *args, **kwargs):
|
|
# Function to generate cache key
|
|
auth_header = None if request.user.is_anonymous else str(request.user.id) if user else None
|
|
custom_path = path if path is not None else request.get_full_path()
|
|
key = generate_cache_key(custom_path, auth_header)
|
|
cached_result = cache.get(key)
|
|
|
|
if cached_result is not None:
|
|
return Response(cached_result["data"], status=cached_result["status"])
|
|
response = view_func(instance, request, *args, **kwargs)
|
|
if response.status_code == 200 and not settings.DEBUG:
|
|
cache.set(
|
|
key,
|
|
{"data": response.data, "status": response.status_code},
|
|
timeout,
|
|
)
|
|
|
|
return response
|
|
|
|
return _wrapped_view
|
|
|
|
return decorator
|
|
|
|
|
|
def invalidate_cache_directly(path=None, url_params=False, user=True, request=None, multiple=False):
|
|
if url_params and path:
|
|
path_with_values = path
|
|
# Assuming `kwargs` could be passed directly if needed, otherwise, skip this part
|
|
for key, value in request.resolver_match.kwargs.items():
|
|
path_with_values = path_with_values.replace(f":{key}", str(value))
|
|
custom_path = path_with_values
|
|
else:
|
|
custom_path = path if path is not None else request.get_full_path()
|
|
auth_header = None if request and request.user.is_anonymous else str(request.user.id) if user else None
|
|
key = generate_cache_key(custom_path, auth_header)
|
|
|
|
if multiple:
|
|
cache.delete_many(keys=cache.keys(f"*{key}*"))
|
|
else:
|
|
cache.delete(key)
|
|
|
|
|
|
def invalidate_cache(path=None, url_params=False, user=True, multiple=False):
|
|
def decorator(view_func):
|
|
@wraps(view_func)
|
|
def _wrapped_view(instance, request, *args, **kwargs):
|
|
# invalidate the cache
|
|
invalidate_cache_directly(
|
|
path=path,
|
|
url_params=url_params,
|
|
user=user,
|
|
request=request,
|
|
multiple=multiple,
|
|
)
|
|
return view_func(instance, request, *args, **kwargs)
|
|
|
|
return _wrapped_view
|
|
|
|
return decorator
|