Wednesday, September 22, 2021

TLRU Cache in Python

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__(selffunc=Nonemaxsize=128ttl=120):
        self.func = func
        self.maxsize = maxsize
        self.ttl = ttl
        self.cache = OrderedDict()
        self.decorator_with_parameters = func == None

    def get_value(selfkey):
        result = self.cache.get(key)
        if not result:
            return None

        valueexpires_at = result

        if expires_at < time.time():
            del self.cache[key]
            return None

        self.cache.move_to_end(key)
        return value


    def put_value(selfkeyvalue):
        if len(self.cache) >= self.maxsize:
            self.cache.popitem(False)
        self.cache[key] = (valuetime.time() + self.ttl)

    def _call(self, *args):
        value = self.get_value(args)
        if value:
            return value

        value = self.func(*args)
        self.put_value(argsvalue)

        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(selfkey):
        result = self.cache.get(key)
        if not result:
            return None

        valueexpires_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(selfkeyvalue):
        if len(self.cache) >= self.maxsize:
            self.cache.popitem(False)
        self.cache[key] = (valuetime.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(argsvalue)

        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__(selffunc=Nonemaxsize=128ttl=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(vy):
    print("Calculating"vy)
    return v * y

Or with parameters:

@TLRU(maxsize=10000ttl=360)
def calc(vy):
    print("Calculating"vy)
    return v * y