Implementing an LRU cache in Python is quite easy, you just use the @lru_cache decorator from the functools library. For TLRU, where items have an expiry time, no standard exists. So here is my implementation:
from collections import OrderedDict
import time
class TLRU:
def __init__(self, func=None, maxsize=128, ttl=120):
self.func = func
self.maxsize = maxsize
self.ttl = ttl
self.cache = OrderedDict()
self.decorator_with_parameters = func == None
def get_value(self, key):
result = self.cache.get(key)
if not result:
return None
value, expires_at = result
if expires_at < time.time():
del self.cache[key]
return None
self.cache.move_to_end(key)
return value
def put_value(self, key, value):
if len(self.cache) >= self.maxsize:
self.cache.popitem(False)
self.cache[key] = (value, time.time() + self.ttl)
def _call(self, *args):
value = self.get_value(args)
if value:
return value
value = self.func(*args)
self.put_value(args, value)
return value
def __call__(self, *args):
if self.decorator_with_parameters:
self.func = args[0]
return self._call
return self._call(*args)
def clear_cache(self):
self.cache.clear()
Now for an explanation of the code, for those interested.
Our cache implementation is using the OrderedDict class from the collections lib. It behaves as a dict, but keeps the items sorted in the order of insertion.
Here is the implementation of the value retrieval method:
def get_value(self, key):
result = self.cache.get(key)
if not result:
return None
value, expires_at = result
if expires_at < time.time():
del self.cache[key]
return None
self.cache.move_to_end(key)
return value
It first looks for the value in the cache. If it does not find it, it returns None immediately. Items in our cache are tuples packing both the value and the expiry time in seconds. So our second action is to check if the value expired. If yes, we remove it from the cache and answers back that we do not have it. If we do find that the value is fresh enough, we push it to the bottom of our OrderedDict using the move_to_end method.
Storing a value goes like this:
def put_value(self, key, value):
if len(self.cache) >= self.maxsize:
self.cache.popitem(False)
self.cache[key] = (value, time.time() + self.ttl)
We do two things here. First, if we reached the cache's maximum size, we remove the oldest value. The False parameter to the popitem method allows the OrderedDict to behave like a FIFO queue. Then, we compute the expiry date and pack it together with the value inside our cache.
Here is the method that wraps our cached function:
def _call(self, *args):
value = self.get_value(args)
if value:
return value
value = self.func(*args)
self.put_value(args, value)
return value
It looks for our value in the cache. If we find it, we just return it. In the other case, we call the wrapped function, store the result into the cache before returning it.
Now for the trickier parts. First the constructor:
def __init__(self, func=None, maxsize=128, ttl=120):
self.func = func
self.maxsize = maxsize
self.ttl = ttl
self.cache = OrderedDict()
self.decorator_with_parameters = func == None
It doesn't look tricky, but you have to know that decorators have two ways of working, depending if you use it with parameters or not:
- Decorators without parameters call the constructor with the wrapped function as the first parameter.
- Decorators with parameters call the constructor without parameters.
That is why the func parameter is optional. We detect which case is running by checking if func is None.
Now for the function called by the decorator pattern:
def __call__(self, *args):
if self.decorator_with_parameters:
self.func = args[0]
return self._call
return self._call(*args)
Again, two cases:
- Decorators without parameters call this function with the wrapped function's parameters, expecting the value as a result
- Decorators with parameters call this function with the wrapped function's reference, expecting a reference to the wrapper as a result.
So if we have parameters in our decorator, we store the wrapped function, which is our first method argument, into the func attribute, then return the wrapper reference. In the other case, we simply call the wrapper immediately.
You can use it like this:
@TLRU
def calc(v, y):
print("Calculating", v, y)
return v * y
Or with parameters:
@TLRU(maxsize=10000, ttl=360)
def calc(v, y):
print("Calculating", v, y)
return v * y
