How to write decorator for Redis Cache

When started looking at one of the business use case where to use either the existing extensions or create own for Django Rest Framework.

So went through the below strategies for redis cache implementation either one of it due to it’s rich support -

django-redis: This is mostly used for backend.

OR

cacheops: This is not supported backend. Mostly used for frontend caching.

After implemented through django-redis, we got the cache eviction typical scenario where the whole application impacted.

And at the time of implementation we used general utility function to get, set and delete from the AWS redis.

So as a next step, i started creating decorator for that. Here are small snippet of that:

Cache_decorator.py

from functools import wraps
from time import perf_counter
import redis
import json

cache = redis.Redis(host=’<redis endpoint>’,port=<redis port>,password=’<redis password>’)

def test_redis():
cache.set(‘foo’, ‘bar’)
value = cache.get(‘foo’)
print(value)

def cached(fn):
“””Decorator that caches the results of the function call.”””

@wraps(fn)
def wrapper(self, req_url, identifier, *args, **kwargs):
# Get the cache key from the function’s arguments.
cache_key = get_cache_key_by_id(req_url, identifier)
result = cache.get(cache_key)
if result is None:
# Run the function and cache the result for next time.
value = fn(self, req_url, identifier, *args, **kwargs)
value_json = json.dumps(value)
cache.set(cache_key, value_json, 120)
else:
# Skip the function entirely and use the cached value instead.
value_json = result.decode(‘utf-8’)
value = json.loads(value_json)
return value
return wrapper

def get_cache_key(fn, *args):
key_parts = [fn.__name__] + list(args)
key = ‘-’.join(key_parts)

return key

cache_test.py

import cache_decorator
import json
import requests

base_url = “https://jsonplaceholder.typicode.com/"
# req_url = “https://jsonplaceholder.typicode.com/todos"
# req_url = “https://jsonplaceholder.typicode.com/todos/2"
req_url = base_url+”todos?userId=10&completed=true”
# req_url = “https://jsonplaceholder.typicode.com/users"

class CacheTesting:

@cache_decorator.cached
def get_deal(self, req_url, identifier, *args, **kwargs):
response = requests.get(req_url)
todos = json.loads(response.text)

print(f”View values from api {todos}”)

return todos

Testing:

t = CacheTesting()
t.get_deal(req_url, “abc123-trtrrdh-4jdn”)