Caching in Python with Redis + @Decorator

TL;DR: Here’s a simple Python decorator I wrote which can cache function results with Redis.


# 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:

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=''), 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
            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
  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 🙂