Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Redis-backed LRU cache decorator in Python.
#!/usr/bin/env python
__author__ = 'Kevin Warrick'
__email__ = ''
import cPickle
from functools import wraps
def redis_lru(capacity=5000, slice=slice(None)):
Simple Redis-based LRU cache decorator *.
*conn* Redis connection
*capacity* maximum number of entries in LRU cache
*slice* slice object for restricting prototype args
Usage is as simple as prepending the decorator to a function,
passing a Redis connection object, and the desired capacity
of your cache.
def func(foo, bar):
# some expensive operation
return baz
Uses 4 Redis keys, all suffixed with the function name:
lru:keys: - sorted set, stores hash keys
lru:vals: - hash, stores function output values
lru:hits: - string, stores hit counter
lru:miss: - string, stores miss counter
* Functions prototypes must be serializable equivalent!
def decorator(func):
cache_keys = "lru:keys:%s" % (func.__name__,)
cache_vals = "lru:vals:%s" % (func.__name__,)
cache_hits = "lru:hits:%s" % (func.__name__,)
cache_miss = "lru:miss:%s" % (func.__name__,)
lvars = [None] # closure mutable
def add(key, value):
conn = lvars[0]
conn.hset(cache_vals, key, cPickle.dumps(value))
conn.zadd(cache_keys, 0, key)
return value
def get(key):
conn = lvars[0]
value = conn.hget(cache_vals, key)
if value:
conn.zincrby(cache_keys, key, 1.0)
value = cPickle.loads(value)
return value
def eject():
conn = lvars[0]
count = min((capacity / 10) or 1, 1000)
if conn.zcard(cache_keys) >= capacity:
eject = conn.zrange(cache_keys, 0, count)
conn.zremrangebyrank(cache_keys, 0, count)
conn.hdel(cache_vals, *eject)
def wrapper(*args, **kwargs):
conn = lvars[0]
if conn:
items = args + tuple(sorted(kwargs.items()))
key = cPickle.dumps(items[slice])
return get(key) or add(key, func(*args, **kwargs))
return func(*args, **kwargs)
def info():
conn = lvars[0]
size = int(conn.zcard(cache_keys) or 0)
hits, misses = int(conn.get(cache_hits) or 0), int(conn.get(cache_miss) or 0)
return hits, misses, capacity, size
def clear():
conn = lvars[0]
conn.delete(cache_keys, cache_vals)
conn.delete(cache_hits, cache_miss)
def init(conn):
lvars[0] = conn
wrapper.init = init = info
wrapper.clear = clear
return wrapper
return decorator
if __name__ == "__main__":
import redis
conn = redis.StrictRedis()
def test(foo, bar, baz=None):
print 'called test.'
return True
conn = redis.StrictRedis()
test(1, 2, baz='A')
test(3, 4, baz='B')
test(1, 2, baz='A')
print "hits %d, misses %d, capacity %d, size %d" %
test(1, 2, baz='A')
test(3, 4, baz='B')
test(1, 2, baz='A')
print "hits %d, misses %d, capacity %d, size %d" %
Copy link

abulka commented Jun 7, 2021

Thanks for this - I forked it and converted it to Python 3

  • Python 3 compatible
  • redis-py 3.0 compatible
  • Made some things more like Python 3 functools.lru_cache
    • renamed .clear() to .cache_clear()
    • renamed .info() to .cache_info()
    • .cache_info() now returns namedtuple object like Python 3 functools.lru_cache does
    • renamed redis_lru capacity parameter to maxsize, allow it to be None
    • enable passing in conn via the decorator
  • Added some asserts to the tests, test a second function

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment