TL;DR: Here’s a simple Python decorator I wrote which can cache function results with Redis.
Prerequisites
# python 3.11, tested on Mac OS X and Ubuntu Linux # installing redis-py pip install --user redis # running a redis container for local development # it's also ok to run redis directly docker run --rm -p 6379:6379 redis:7.0-alpine
The Python Decorator
What’s a decorator in Python? Here’s a very easy-to-understand explanation just for the question. Below is my code of the decorator with notes:
# redisHelper.py
import redis
import os
# I always go for connection pooling whenever it's available
redis_pool = redis.ConnectionPool(host=os.environ.get("REDIS_HOST", default='127.0.0.1'), port=6379, db=0)
# I use the function name + all parameters to form a cache key
# so whatever combination of values used to call a funtion will have its unique cache key
# eg. cache key for myfunc('abc', 123) will be 'myfunc_abc_123'
def args_to_key(*args, **kwargs):
params = [arg.__name__ if callable(arg) else str(arg) for arg in args] + [str(kwarg) for kwarg in kwargs.values()]
return "_".join(params)
# the name of the decorator
def redis_cached(func):
def wrapper(*args, **kwargs):
# reuse a connection from the pool
r = redis.Redis(connection_pool=redis_pool)
cache_key = args_to_key(func, *args, **kwargs)
# test if a matching cache key exists
cached = r.get(cache_key)
if cached:
# found in cache, return it
# redis returns bytes, converted to string here
return cached.decode('utf-8')
# otherwise pass everything to the downstream function
result = func(*args, **kwargs)
# set cache time-to-live duration if there's a ttl parameter
if 'ttl' in kwargs:
ttl_seconds = kwargs['ttl']
# default ttl is 1 hour
# after the ttl, the key will be removed from redis automatically
else:
ttl_seconds = 3600
# put the result from downstream function into cache, with a ttl
# so next call with the same parameters will be handled by the cache
r.setex(cache_key, ttl_seconds, result)
# return the result transparently
return result
return wrapper
# sample usage
@redis_cached
def test_redis(key1, key2):
# the function body could be some database queries which worth caching
return f"a string contains {key1} and {key2}"Done 🙂