Expirable LRU cache

Question

The requirements are:

• evict any expired items
• find the items with the lowest priority and evict the least recently used one

Any feedback would be appreciated.

from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity):
        self.capacity = capacity
        self.cache = OrderedDict() 
        self.timer = 0 

    def get(self, key):
        if key in self.cache:
            value, timestamp = self.cache[key]
            if self.timer - timestamp <= value[1]: 
                self.cache.pop(key)
                self.cache[key] = (value, self.timer) 
                self.timer += 1
                return value[0]
            else:
                self.cache.pop(key)
        return -1

    def put(self, key, value, ttl):
        if key in self.cache:
            self.cache.pop(key)
        elif len(self.cache) >= self.capacity:
            self.cleanup(ttl)

        if len(self.cache) >= self.capacity:
            self.cache.popitem(last=False)

        self.cache[key] = ((value, ttl), self.timer)
        self.timer += 1

    def cleanup(self, ttl):
        expired_keys = []
        min_priority = float('inf')
        print(min_priority)
        min_priority_keys = []

        for key, ((priority, ttl), timestamp) in self.cache.items():
            print(key, ((priority, ttl), timestamp), self.timer)
            if self.timer - timestamp > ttl:
                expired_keys.append(key)

            if priority < min_priority:
                min_priority = priority
                min_priority_keys = [key]
            elif priority == min_priority:
                min_priority_keys.append(key)

        if expired_keys:
            for key in expired_keys:
                self.cache.pop(key)
        else:
            min_lru = float('inf')
            min_lru_key = None

            for key, (_, timestamp) in self.cache.items():
                if key in min_priority_keys and timestamp < min_lru:
                    min_lru = timestamp
                    min_lru_key = key

            if min_lru_key is not None:
                self.cache.pop(min_lru_key)


# Usage example:
cache = LRUCache(3)
cache.put("key1", 3, 2)
cache.put("key2", 5, 20)
cache.put("key3", 1, 20)
cache.put("key4", 10, 29)


# # After updating the timer, retrieve items, and expired items will be automatically removed
print(cache.get("key1")) # is expired
cache.put("key5", 5, 29)
print(cache.get("key2")) # has higher priority than key3
print(cache.get("key3")) # was evicted to make space for key 5 -> (key4, key5, key2)

Answer 1

name

A lru_cache with expiry, nice! I like the requirements.

Calling it LRUCache seems to invite confusion with the well-known functools decorator, at least during telephone discussions. Consider calling it a "timed LRU cache", highlighting the TTL field.

Consider also offering its functionality via a @decorator.

---

default TTL

Time to live is a mandatory argument on each .put() call.

    def __init__(self, capacity):

Consider appending a ... , default_ttl: float): argument to the __init__ ctor, which would let the .put() TTL be optional.

Type hints in app-level code are strictly optional, added when the application author feels they are helpful. But a library author has a broader set of responsibilities to diverse callers. It would be a kindness to type hint this code so mypy lints cleanly. Notice that PEP 484 offers an "int TTLs are fine in a float context" shortcut. The capacity clearly should be an int.

I have maintained a fair amount of redis client application code, which inevitably stores a bunch of keys with some TTL value. In a given app the TTL might be constant or it might vary across key categories. I can tell you that it is much easier for a maintainer to gain an understanding of what is going on, what invariants will hold, if TTL behavior is "boring". Often this would be done by defining a MANIFEST_CONSTANT which conventionally appears in each request. It would be very helpful if you offer a "path of least resistance" to support this common need of authors, for clearly expressing their intent, and of maintainers, for rapidly understanding app behavior.

---

LRU on update

    def put(self, key, value, ttl):
        if key in self.cache:
            self.cache.pop(key)

This is perfectly nice; it is an interesting design choice. As a library user it wouldn't be obvious to me that storing X under key1 and later updating it to Y would make key1 least recently used. It could go either way. Just add a docstring that describes the behavior. (Note that, since you elected to call it public cache rather than private _cache, caller could always .pop() if desired.)

The rest of the logic is very clear, thank you.

And then we come to this:

        self.timer += 1

What? Time to live does not have units of wallclock seconds?!? Ok, we really need to have a talk about writing """docstrings""". It is necessary to tell callers what your library does. In this case, what seemed to be clear names weren't. The naming obscures the actual behavior. The semantics feels a bit like "each router hop shall decrement the IP TTL field", event-based rather than time-based.

---

tidy up before review

    def cleanup(self, ttl):
        ...
        print(min_priority)
            ...
            print(key, ((priority, ttl), timestamp), self.timer)

You're requesting review prior to merging down to main. This is the right time to delete debug logging and chatty print() statements.

If you need instrumentation, consider incrementing counters instead.

---

time complexity

After a warmup period, typically every .put() will do a .cleanup().

        for key, ((priority, ttl), timestamp) in self.cache.items():

Ummm, we linearly scan the entire cache? Every time?!?

The cleanup docstring really needs to warn about that. And / or we need to find a way to bail early, or amortize the cost. For example we might admit a "double capacity" internal memory footprint, and let .get() discard old entries which should not be externally visible. That would reduce the need for every single put to do a full cleanup.

I was really expecting to find a TTL priority queue at the center of this library. With capacity C, that would let you discard K old entries in K × log(C) time.

It might be helpful to have a very short cleanup which calls a pair of helpers. Then the docstring for each could explain the single responsibility being addressed, either expiring old entries in a LRU sense or in a TTL sense.

---

memory complexity

        self.cache[key] = ((value, ttl), self.timer)

This is an interesting storage scheme. I like it, feel free to keep it as-is.

But the pair of tuples requires one extra pointer for that one extra tuple. So an app that is storing small integer keys and values will pay noticeably more overhead compared to a simpler scheme.

Despite the convenience of certain unpack operations in cleanup, consider converting this to just a single 3-tuple, to save memory.

---

unit tests

print(cache.get("key1")) # is expired

Thank you for offering example usage.

This is not a self-evaluating test; it does not know the right answer. Rather, it invites a human to eyeball the answer and make a decision.

Please add a proper test suite. For example, show the expiration with self.assertEqual(-1, cache.get("key1")).

---

sentinel

    def get(self, key):
        ...
        return -1

This is astonishing behavior.

Minimally, describe it in a docstring.

Suppose an app stores integers in the range -3 .. 3. It has no way to distinguish "missing" from a legitimate value. Consider relying on the usual sentinel_missing = object() idiom to invent a value that caller definitely won't attempt to store.

Or offer an optional ... , default=None): parameter as other APIs do. Or raise KeyError.

tuple unpack

            value, timestamp = self.cache[key]

Rather than use obscure [0] / [1] subscripts on value, please use the same ((priority, ttl), timestamp) unpack that cleanup does.

---

invariants

The high level advice from this review is that the source code really needs to document its behavior.

The class docstring should introduce the concept behind self.timer, distinguishing it from elapsed wallclock time.

Each method docstring should describe its complexity along with app-level behavior. Assume a user will cache some thousands of entries, and that "warmup" behavior before capacity is reached is of relatively little interest.

Invariants of interest to a caller include statements like "after this call all TTLs will be in [this] range" or "old stored values have expired after [this] or [that] happens".

---

This library achieves a subset of its design goals. Scaling and documentation are areas of concern.

I would be willing to delegate or accept maintenance tasks on this codebase.