How to Build an LRU Cache From Scratch in Python to Bound Memory Without Losing Hot Data
Build a real LRU cache from scratch in Python, verify it with a randomized fuzz test, add TTL expiry, prove and fix a genuine thread-safety race, and benchmark why the data structure choice matters...
An LRU cache, short for “least recently used,” is one of the most common pieces of infrastructure in software you use every day: a database driver caching query results, a web framework caching rendered pages, an operating system caching disk blocks in RAM, even Python’s own functools.lru_cache decorator. The idea sounds simple: keep a fixed number of the most recently accessed items, and when a new item needs room, throw out whichever one has gone the longest without being touched. In this tutorial you will build one from scratch in Python, verify it against a reference implementation with a randomized fuzz test that catches a real bug along the way, add time-based expiry, make it safe for concurrent use by proving (not assuming) that a race condition exists and then fixing it, and measure exactly how much the right data structure choice matters at scale. Every piece of code and every number in this post comes from code that was actually run, including two mistakes this tutorial’s own author made and caught while building it.
Table Of Content
- What an LRU Cache Is, and Why It Matters
- Prerequisites
- Step 1: The Problem With a Cache That Never Forgets
- Step 2: Bounding It the Naive Way, and Paying for It
- Step 3: O(1) Eviction With OrderedDict
- Step 4: The Bug Almost Everyone Writes: Forgetting to Touch on Read
- Step 5: Building the Real Thing, a Doubly Linked List Plus a Hash Map
- Proving It Matches, and Catching a Real Bug in the Process
- Step 6: Adding Time-Based Expiry (TTL)
- The Gotcha: “Lazy” Expiry Means Dead Entries Sit There Until Something Notices
- Step 7: Making It Thread-Safe
- Step 8: When to Just Use functools.lru_cache Instead
- Step 9: Proving the O(1) Claim With a Benchmark
- Common Mistakes and Gotchas
- How to Verify It All Works End to End
- Next Steps
What an LRU Cache Is, and Why It Matters
A cache is just a smaller, faster copy of some data that is slower or more expensive to fetch: a network call, a disk read, an expensive computation. Caches only help if they stay small enough to actually be fast, which means something has to decide what to keep and what to throw away once the cache is full. “Least recently used” is a simple, effective answer to that question, based on a reasonable real-world assumption called temporal locality: if something was accessed recently, it is more likely to be accessed again soon than something nobody has touched in a while.
Two operations define an LRU cache:
- get(key): return the value for a key if it is present, and mark it as just used.
- put(key, value): store a value for a key, marking it as just used. If the cache is now over capacity, evict whichever key has gone the longest without being used.
The interesting engineering problem is not “how do I remember what was used least recently,” it is “how do I do that without get and put becoming slower as the cache grows.” That is exactly what this tutorial walks through, one honest mistake at a time.
Prerequisites
- Python 3.10 or newer (this tutorial was built and tested on Python 3.13.14, using only the standard library: no packages to install for the core cache itself)
pytestfor the final verification suite (pip install pytest; tested against pytest 9.1.1)- Basic familiarity with Python classes, dictionaries, and what “O(1)” versus “O(n)” time complexity means in general terms
- A terminal and a working directory you can throw away afterward
Step 1: The Problem With a Cache That Never Forgets
Start with the simplest possible thing that could be called a cache: a dictionary with a get and a put method.
class NaiveCache:
"""A 'cache' that never forgets anything."""
def __init__(self):
self.store = {}
def get(self, key):
return self.store.get(key)
def put(self, key, value):
self.store[key] = value
if __name__ == "__main__":
cache = NaiveCache()
for i in range(5):
cache.put(f"user:{i}", {"name": f"User {i}"})
print(f"Cache size after 5 puts: {len(cache.store)}")
for i in range(100_000):
cache.put(f"user:{i}", {"name": f"User {i}"})
print(f"Cache size after 100,000 puts: {len(cache.store)}")
print(f"user:0 still present: {cache.get('user:0') is not None}")
Run it, and the problem is immediate:
$ python step1_naive_cache.py
Cache size after 5 puts: 5
Cache size after 100,000 puts: 100000
user:0 still present: True
This “cache” holds everything forever. In a real service, that is a slow memory leak with a friendly name. A cache needs a capacity, and it needs a rule for what to remove once that capacity is reached.
Step 2: Bounding It the Naive Way, and Paying for It
The obvious fix: keep a capacity, and track access order in a plain list, moving a key to the end every time it is touched, so the front of the list is always the least recently used entry.
class ListLRUCache:
"""Correct LRU behavior, but every get/put touches the whole order list."""
def __init__(self, capacity):
self.capacity = capacity
self.store = {}
self.order = [] # index 0 = least recently used, end = most recently used
def get(self, key):
if key not in self.store:
return None
self.order.remove(key) # O(n) scan
self.order.append(key)
return self.store[key]
def put(self, key, value):
if key in self.store:
self.order.remove(key) # O(n) scan
elif len(self.store) >= self.capacity:
oldest = self.order.pop(0) # O(n) shift
del self.store[oldest]
self.store[key] = value
self.order.append(key)
This is correct. Watch it evict the right entry:
$ python step2_list_lru.py
After filling a,b,c: order=['a', 'b', 'c']
After get('a'): order=['b', 'c', 'a']
After put('d',4): order=['c', 'a', 'd'], store keys=['a', 'c', 'd']
get('b') after eviction: None
b was the least recently used entry once a got touched by a get(), so b is exactly what gets evicted when d needs room. The logic is right. The problem is cost: list.remove(key) has to scan the list looking for that key, and list.pop(0) has to shift every remaining element down by one slot. Both are O(n) operations, where n is however many items are currently in the cache. A cache holding 50,000 entries pays that full scan cost on every single get and put. Step 9 puts a real number on exactly how bad that gets.
Step 3: O(1) Eviction With OrderedDict
Python’s standard library already has a data structure built for exactly this. Its own documentation is direct about why: “The regular dict was designed to be very good at mapping operations. Tracking insertion order was secondary. The OrderedDict was designed to be good at reordering operations… this makes it suitable for implementing various kinds of LRU caches.” Python’s collections documentation says so explicitly.
Two methods do all the work: move_to_end(key), which repositions an existing key without touching anything else, and popitem(last=False), which removes and returns the oldest entry. Both are backed by a doubly linked list internally, which is exactly what makes them fast regardless of how many entries are in the dictionary, a point Step 5 makes concrete by building that same structure by hand.
from collections import OrderedDict
class LRUCache:
"""O(1) get/put using OrderedDict's move_to_end + popitem."""
def __init__(self, capacity):
self.capacity = capacity
self.store = OrderedDict()
def get(self, key):
if key not in self.store:
return None
self.store.move_to_end(key)
return self.store[key]
def put(self, key, value):
if key in self.store:
self.store.move_to_end(key)
self.store[key] = value
if len(self.store) > self.capacity:
self.store.popitem(last=False)
That is a complete, correct, O(1) LRU cache in sixteen lines of code. For most real applications, this is genuinely all you need. But there is a mistake buried in how easy this looks, and it is worth seeing go wrong on purpose.
Step 4: The Bug Almost Everyone Writes: Forgetting to Touch on Read
The single most common LRU bug is updating recency on put() but forgetting to do it on get() too. It looks harmless, because the cache still returns the right value; it just silently evicts the wrong entry later. Here is the broken version, differing from Step 3 by exactly one missing line:
class BrokenLRUCache:
"""The common bug: only reorders on put(), never on get()."""
def __init__(self, capacity):
self.capacity = capacity
self.store = OrderedDict()
def get(self, key):
if key not in self.store:
return None
return self.store[key] # BUG: no move_to_end here
def put(self, key, value):
if key in self.store:
self.store.move_to_end(key)
self.store[key] = value
if len(self.store) > self.capacity:
self.store.popitem(last=False)
def run_scenario(cache_cls, label):
cache = cache_cls(capacity=3)
cache.put("a", 1)
cache.put("b", 2)
cache.put("c", 3)
cache.get("a") # a was just read; it should now count as "fresh"
cache.put("d", 4) # capacity=3 forces one eviction
a_survived = cache.get("a") is not None
b_survived = cache.get("b") is not None
print(f"{label}: a_survived={a_survived}, b_survived={b_survived}, "
f"keys={list(cache.store.keys())}")
Run both side by side:
$ python step3_ordereddict_lru.py
Correct (move_to_end on get): a_survived=True, b_survived=False, keys=['c', 'd', 'a']
Broken (no move_to_end on get): a_survived=False, b_survived=True, keys=['b', 'c', 'd']
That is a completely wrong eviction, verified, not assumed: a was read a moment before the eviction and still gets thrown away, while b, which nobody touched, survives. If this were a web session cache, that is a user getting logged out right after being active. If it is a per-user rate limiter, that is the wrong user getting throttled. The fix is one line, but you have to remember it is needed, because both versions compile, run, and return correct values on every call; only the eviction order is wrong, and that only shows up under real capacity pressure.
Step 5: Building the Real Thing, a Doubly Linked List Plus a Hash Map
OrderedDict is the right tool for real code, but understanding what it is doing underneath makes the rest of this tutorial easier to reason about, and it is the classic version of this problem for a reason. The trick is combining two data structures, each covering the other’s weakness:
- A hash map (Python
dict) gives O(1) lookup by key, but a plain dict has no cheap way to find or remove “the oldest entry.” - A doubly linked list gives O(1) removal and O(1) insertion at either end once you already have a reference to the node, but finding a node by key would normally mean scanning the whole list.
Put a node’s position in the list, and store a direct reference to that node in the dict, keyed by the cache key. Now you can jump straight to any node in O(1) via the dict, and once you have the node, unlinking it and relinking it at the front of the list is also O(1), with no scanning in either direction.
class _Node:
__slots__ = ("key", "value", "prev", "next")
def __init__(self, key=None, value=None):
self.key = key
self.value = value
self.prev = None
self.next = None
class LRUCacheFromScratch:
"""Hash map (O(1) lookup) + doubly linked list (O(1) reorder) = O(1) LRU.
head.next is the most-recently-used entry; tail.prev is the
least-recently-used entry, the one evicted first.
"""
def __init__(self, capacity):
self.capacity = capacity
self.map = {}
self.head = _Node()
self.tail = _Node()
self.head.next = self.tail
self.tail.prev = self.head
def _unlink(self, node):
node.prev.next = node.next
node.next.prev = node.prev
def _push_front(self, node):
node.next = self.head.next
node.prev = self.head
self.head.next.prev = node
self.head.next = node
def get(self, key):
node = self.map.get(key)
if node is None:
return None
self._unlink(node)
self._push_front(node)
return node.value
def put(self, key, value):
node = self.map.get(key)
if node is not None:
node.value = value
self._unlink(node)
self._push_front(node)
return
if len(self.map) >= self.capacity:
lru_node = self.tail.prev
self._unlink(lru_node)
del self.map[lru_node.key]
node = _Node(key, value)
self.map[key] = node
self._push_front(node)
The two dummy sentinel nodes (head and tail) exist purely to avoid writing special-case code for “the list is empty” or “removing the only remaining node.” Every real node always has a real prev and a real next to link against, even at the boundaries.
Proving It Matches, and Catching a Real Bug in the Process
Reading through linked-list pointer code and convincing yourself it is correct is not the same as proving it. A better test: build a second, independent implementation using the trusted OrderedDict from Step 3, then fire thousands of randomized get and put operations at both in lockstep, and assert that they agree after every single one.
class ReferenceLRUCache:
"""The OrderedDict version from Step 3, used here as a ground truth."""
def __init__(self, capacity):
self.capacity = capacity
self.store = OrderedDict()
def get(self, key):
if key not in self.store:
return None
self.store.move_to_end(key, last=False)
return self.store[key]
def put(self, key, value):
if key in self.store:
self.store.move_to_end(key, last=False)
self.store[key] = value
if len(self.store) > self.capacity:
self.store.popitem(last=True)
The first run of the fuzz test against this reference failed immediately:
$ python step4_scratch_lru.py
Fuzz test passed: False
order mismatch at op 7: scratch=['k129', 'k26'] reference=['k26', 'k129']
The bug was not in the linked-list implementation. It was in this reference class: put() only called move_to_end(key, last=False) inside the if key in self.store: branch, so it correctly re-positioned a key that already existed, but a brand-new key, inserted for the first time, landed wherever Python’s dict naturally puts a new key (the end) and was never explicitly moved to the “most recently used” position at all. The fix has to run for new keys too, not just updates:
def put(self, key, value):
self.store[key] = value
self.store.move_to_end(key, last=False) # must run for NEW keys too
if len(self.store) > self.capacity:
self.store.popitem(last=True)
With that fixed, 20,000 randomized operations produce identical state at every single step:
$ python step4_scratch_lru.py
Fuzz test passed: True
20000 randomized operations produced identical state every step
The lesson generalizes past this one bug: whenever your LRU cache convention is “most recent goes to end X,” every code path that touches a key, including the very first time it is ever inserted, has to honor that convention. It is an easy line to skip, because “put a new key” often does not feel like it needs the same recency-marking step as “touch an existing key.” A randomized cross-check against a second, independently written implementation is a genuinely effective way to catch this class of mistake, because it does not require you to guess which specific scenario is broken in advance.
Step 6: Adding Time-Based Expiry (TTL)
Capacity-based eviction alone is not enough for things like session tokens or one-time codes, which need to disappear after a fixed amount of time even if the cache never fills up. Add a per-entry expiration timestamp, checked lazily on read:
class TTLLRUCache:
"""LRU cache with per-entry TTL, using lazy (read-time) expiry only."""
def __init__(self, capacity, default_ttl=None):
self.capacity = capacity
self.default_ttl = default_ttl
self.store = OrderedDict() # key -> (value, expires_at | None)
def _expired(self, expires_at):
return expires_at is not None and time.monotonic() > expires_at
def get(self, key):
if key not in self.store:
return None
value, expires_at = self.store[key]
if self._expired(expires_at):
del self.store[key] # lazy purge, only triggered by a read
return None
self.store.move_to_end(key)
return value
def put(self, key, value, ttl=None):
ttl = self.default_ttl if ttl is None else ttl
expires_at = time.monotonic() + ttl if ttl else None
if key in self.store:
self.store.move_to_end(key)
self.store[key] = (value, expires_at)
if len(self.store) > self.capacity:
self.store.popitem(last=False)
time.monotonic() is deliberate here instead of time.time(): it cannot jump backward if the system clock gets adjusted (NTP sync, a manual change, daylight saving in some configurations), which matters for anything computing a duration rather than a wall-clock timestamp.
The Gotcha: “Lazy” Expiry Means Dead Entries Sit There Until Something Notices
The word “lazy” is doing real work in that description. An entry is only actually removed from the dict when something calls get() on that exact key after it has expired, or when it eventually reaches the least-recently-used end and gets evicted for space. If neither happens, the dead entry just sits in memory, and len(self.store) keeps reporting it as if it were live data.
def live_size(self):
"""How many entries are NOT expired, without mutating anything."""
return sum(
1 for _value, expires_at in self.store.values()
if not self._expired(expires_at)
)
Simulate a very common real pattern: a burst of short-lived tokens (one-time codes, idempotency keys, rate-limit counters) that get written once and never read again.
$ python step5_ttl_lru.py
After a burst of 500 short-TTL writes: raw size=500
After the TTL window passes, with no reads and no new writes:
len(cache.store) = 500 <- looks 50% full
cache.live_size() = 0 <- actually 0% useful
A dashboard graphing len(cache.store) would show a cache that
never shrinks, even though every entry inside it is dead.
After an explicit sweep_expired(): purged=500, raw size now=0
Two real, practical consequences follow from this. First, if you monitor cache health by graphing raw size against capacity, that graph can quietly lie to you, showing a cache that looks steadily full when it is actually holding nothing useful, which can mask both real capacity problems and real memory waste. Second, if writes are bursty and reads are rare (exactly the one-time-token pattern above), dead entries can sit in memory for a long time, since nothing is forcing them out. The fix is an explicit, periodic sweep_expired() call, run on a timer in a background thread or scheduled task in a real service, rather than trusting lazy expiry alone to reclaim memory on any particular schedule.
Step 7: Making It Thread-Safe
Every version so far assumes one thread at a time. Multiple threads calling put() concurrently on a shared cache is a genuinely common real scenario (a web server handling concurrent requests, a background worker pool), so it is worth checking, not assuming, whether that is actually safe.
Python’s own FAQ is precise about what the GIL does and does not protect: “Each bytecode instruction… is therefore atomic from the point of view of a Python program,” but it gives a direct list of what that does not cover, including D[x] = D[x] + 1, a compound read-modify-write sequence spanning more than one bytecode instruction. The eviction logic in this cache’s put() is exactly that shape: if len(self.store) > self.capacity: self.store.popitem(...) reads the length, then, based on that read, decides whether to act. Between the read and the act, the interpreter can switch to another thread.
In practice, with simple string keys, that gap is narrow enough that a short demo often will not hit it by chance, at default thread-switching settings, even under real concurrent load. Rather than claim a race that cannot actually be shown, this tutorial widens the naturally occurring gap with a small, disclosed, deliberate delay, a standard technique for reliably demonstrating a real but timing-sensitive bug:
class UnsafeSlowLRUCache(LRUCache):
"""Identical logic, with a deliberate delay inside the check-then-act
eviction gap. On a real multi-core machine under heavy load, the GIL
can release between the len() check and the popitem() call; this
class widens that naturally-occurring gap so the race is reliably
visible in a short demo instead of only once in a rare while."""
def _maybe_evict(self):
if len(self.store) > self.capacity:
time.sleep(0.0005) # widen the window between check and act
self.store.popitem(last=False)
class ThreadSafeLRUCache(LRUCache):
"""Same logic, guarded by a single lock around each public method."""
def __init__(self, capacity):
super().__init__(capacity)
self._lock = threading.Lock()
def get(self, key):
with self._lock:
return super().get(key)
def put(self, key, value):
with self._lock:
return super().put(key, value)
Twelve threads, each doing 40 puts against a capacity-20 cache with plenty of eviction pressure:
$ python step6_thread_safety.py
12 threads x 40 puts each, capacity=20
UnsafeSlowLRUCache (widened check-then-act eviction gap):
final size = 13 (capacity is 20)
exceptions raised = 0
ThreadSafeLRUCache (same widened gap, but under a lock):
final size = 20 (capacity is 20)
exceptions raised = 0
No exception is raised in either case, which is itself worth noting: this bug does not announce itself with a crash. The unsafe cache silently ends up under its stated capacity (13 instead of 20), because multiple threads can each observe “we are over capacity” based on a stale length and each independently evict an entry, over-evicting relative to what was actually necessary. That is wasted cache space and unnecessary re-fetches of data that should still have been cached. Wrapping each public method body in a single threading.Lock makes the whole check-then-act sequence atomic again, and the cache lands exactly on capacity every time. The lock does cost some throughput under contention, which is the honest tradeoff: correctness under concurrent access, in exchange for one lock acquisition per call.
Step 8: When to Just Use functools.lru_cache Instead
Python ships a decorator that solves a large, common subset of this problem in one line, and for a lot of real code it is the right answer. Python’s functools documentation describes it as a “decorator to wrap a function with a memoizing callable that saves up to the maxsize most recent calls.”
from functools import lru_cache
@lru_cache(maxsize=128)
def slow_square(n):
time.sleep(0.05) # simulate an expensive computation
return n * n
start = time.perf_counter()
result = slow_square(7)
first_call = time.perf_counter() - start
start = time.perf_counter()
result = slow_square(7)
second_call = time.perf_counter() - start
$ python step7_functools_comparison.py
First call: slow_square(7) = 49, took 0.0509s
Second call: slow_square(7) = 49, took 0.0000s (cached)
cache_info(): CacheInfo(hits=1, misses=1, maxsize=128, currsize=1)
Has a way to evict a single key without clearing everything? False
After cache_clear(): CacheInfo(hits=0, misses=0, maxsize=128, currsize=0)
Calling with an unhashable argument raises: TypeError: unhashable type: 'list'
Reach for functools.lru_cache when you are caching the return value of a pure-ish function of its arguments, inside a single process. It is genuinely O(1), built into the standard library, and requires zero code of your own. It has real limits worth knowing before you hit them in production: the documentation confirms “the positional and keyword arguments to the function must be hashable,” so it cannot cache calls keyed on a list, dict, or other mutable argument; there is no built-in TTL; and there is no way to invalidate a single key, only cache_clear(), which wipes everything. Reach for a hand-built cache like the one in this tutorial when you need TTL, per-key invalidation, size introspection beyond a bare count, or you are caching something that is not simply “the output of this one function.”
Step 9: Proving the O(1) Claim With a Benchmark
“O(1) versus O(n)” is a claim, not a fact, until it is measured. The first version of this benchmark actually produced a confusing, wrong-looking result: at very large capacity, the naive list-based cache looked faster than the O(1) version. The reason was a real flaw in the benchmark’s own design, not in either cache: the key space had grown alongside the capacity, so most operations were inserting brand-new keys that never triggered an eviction or a reorder, meaning the list-based cache’s expensive code paths were barely being exercised at all. A benchmark that does not actually stress the thing it claims to measure is worse than no benchmark, so here is the corrected version, which pre-fills each cache to capacity first, guaranteeing every subsequent operation causes a real eviction or a real reorder:
def benchmark(cache_cls, capacity, operations, key_space):
cache = cache_cls(capacity)
# Pre-fill to capacity so every subsequent op causes real churn
# (an eviction or a reorder of an existing entry), not a free insert.
for i in range(capacity):
cache.put(f"key{i}", i)
start = time.perf_counter()
for i in range(operations):
key = f"key{i % key_space}"
if i % 3 == 0:
cache.get(key)
else:
cache.put(key, i)
return time.perf_counter() - start
$ python step8_benchmark.py
Each run pre-fills the cache to capacity, then performs 15000 get/put ops against a key space 1.5x the capacity, so the cache stays full and churns on every op.
capacity= 500: list=0.0024s dict=0.0027s speedup=0.9x
capacity= 2000: list=0.0371s dict=0.0026s speedup=14.3x
capacity= 8000: list=0.1669s dict=0.0030s speedup=55.5x
capacity= 16000: list=0.3912s dict=0.0027s speedup=147.0x
At a capacity of 500, the two are roughly tied; the fixed overhead of Python function calls dominates at that size, and a 500-element list scan is cheap enough not to matter yet. From there the picture changes fast: by 16,000 entries, the O(1) version is 147 times faster, and the dict-based version’s own time barely moves across the whole range (0.0026s to 0.0030s) while the list-based version’s time grows in direct proportion to capacity, exactly what O(n) per operation predicts. That crossover point, where the naive approach stops being “fine” and starts being a real problem, is worth remembering: it is not that the naive version is always bad, it is that it gets worse in a way that is easy to miss in development with small test data and painful to discover in production with real data.
Common Mistakes and Gotchas
- Forgetting to mark an entry as recently used on
get(), not justput(). Step 4 reproduced this exactly: the cache still returns correct values, but silently evicts the wrong entry later. Both read and write paths must update recency. - Forgetting a brand-new key needs the same recency update as an updated one. Step 5’s fuzz test caught this for real: a new key inserted via plain dict assignment lands wherever the dict naturally places it, not necessarily where your “most recently used” convention expects it.
- Trusting
len(cache)as a measure of useful, live data when TTL is involved. Step 6 showed a cache that reported itself as half full while holding zero live entries, because lazy expiry only removes an entry when something reads that specific key or eviction pressure reaches it. - Assuming single-op atomicity covers a multi-step sequence. Step 7 showed that “check the size, then maybe evict” is not atomic just because each individual dict operation inside it is. Any compound read-then-act sequence on shared state needs an explicit lock under concurrent access.
- Reaching for a hand-built cache when
functools.lru_cachewould do. If you are only memoizing a pure function’s return value within one process and do not need TTL or per-key eviction, the one-line decorator from Step 8 is less code to maintain and just as fast.
How to Verify It All Works End to End
The final, combined cache module folds together capacity-based eviction, TTL, and thread safety into one class, and adds three small pieces that were implied but not yet written out: delete() for explicit single-key removal, sweep_expired() from Step 6 for actively clearing dead entries, and __len__/__contains__ so the cache works with Python’s built-in len() and in:
import threading
import time
from collections import OrderedDict
_MISSING = object()
class LRUCache:
"""A thread-safe, O(1) LRU cache with optional per-entry TTL.
- get/put are O(1) via OrderedDict's move_to_end + popitem.
- Every entry that is read updates its recency, including reads
that discover the entry has expired (which purge it).
- Expired entries are only removed lazily (on get) or when eviction
is needed on put; call sweep_expired() periodically in a real
service if entries are written once and rarely re-read.
"""
def __init__(self, capacity, default_ttl=None):
if capacity <= 0:
raise ValueError("capacity must be positive")
self.capacity = capacity
self.default_ttl = default_ttl
self._store = OrderedDict() # key -> (value, expires_at | None)
self._lock = threading.Lock()
def _expired(self, expires_at):
return expires_at is not None and time.monotonic() > expires_at
def get(self, key, default=None):
with self._lock:
entry = self._store.get(key)
if entry is None:
return default
value, expires_at = entry
if self._expired(expires_at):
del self._store[key]
return default
self._store.move_to_end(key)
return value
def put(self, key, value, ttl=None):
ttl = self.default_ttl if ttl is None else ttl
expires_at = time.monotonic() + ttl if ttl else None
with self._lock:
self._store[key] = (value, expires_at)
self._store.move_to_end(key)
if len(self._store) > self.capacity:
self._store.popitem(last=False)
def delete(self, key):
with self._lock:
self._store.pop(key, None)
def sweep_expired(self):
with self._lock:
dead = [k for k, (_v, exp) in self._store.items() if self._expired(exp)]
for k in dead:
del self._store[k]
return len(dead)
def __len__(self):
with self._lock:
return len(self._store)
def __contains__(self, key):
return self.get(key, _MISSING) is not _MISSING
Beyond eyeballing printed output, write the guarantees down as tests so a future change to this code cannot silently break any of them:
def test_eviction_order_respects_reads():
cache = LRUCache(capacity=2)
cache.put("a", 1)
cache.put("b", 2)
cache.get("a") # a is now more recently used than b
cache.put("c", 3) # capacity=2 forces an eviction -> b should go
assert cache.get("a") == 1
assert cache.get("b") is None
assert cache.get("c") == 3
def test_never_exceeds_capacity_under_concurrent_load():
cache = LRUCache(capacity=25)
errors = []
def worker(thread_id):
try:
for i in range(500):
cache.put(f"t{thread_id}-{i}", i)
except Exception as exc:
errors.append(exc)
threads = [threading.Thread(target=worker, args=(t,)) for t in range(16)]
for t in threads:
t.start()
for t in threads:
t.join()
assert not errors
assert len(cache) == 25
Running the full suite (ten tests covering basic get/put, eviction order, recency on updates, TTL expiry with and without a default, sweeping dead entries, deletion, containment checks, capacity validation, and sixteen threads hammering the cache concurrently) against this exact code:
$ python -m pytest test_cache.py -v
test_cache.py::test_basic_get_put PASSED [ 10%]
test_cache.py::test_eviction_order_respects_reads PASSED [ 20%]
test_cache.py::test_update_existing_key_refreshes_recency PASSED [ 30%]
test_cache.py::test_ttl_expiry PASSED [ 40%]
test_cache.py::test_ttl_override_per_key PASSED [ 50%]
test_cache.py::test_sweep_expired_purges_dead_entries PASSED [ 60%]
test_cache.py::test_delete PASSED [ 70%]
test_cache.py::test_contains PASSED [ 80%]
test_cache.py::test_capacity_zero_rejected PASSED [ 90%]
test_cache.py::test_never_exceeds_capacity_under_concurrent_load PASSED [100%]
============================= 10 passed in 0.36s ==============================
Ten tests, ten passes, including the exact concurrent-load scenario Step 7 proved was broken without the lock.
Next Steps
From here, a few directions are worth exploring depending on what you are actually building:
- If your cache needs to survive a process restart, see How to Build a Write-Ahead Log in Python to Survive a Crash Without Losing Data for the same “log first, apply second” discipline, applied to durable, crash-safe writes.
- If a single process’s cache is not enough because you are running many instances behind a load balancer, the next problem is a distributed cache (Redis and similar tools solve this), plus deciding how to keep multiple caches roughly consistent with each other.
- If you are caching results from an LLM specifically, see How to Build a Semantic Cache for LLM Calls With Ollama and Python, which covers a related but different problem: matching semantically similar (not identical) requests, using embeddings and cosine similarity instead of exact key matching.
- If cache misses are expensive and frequent at the same time (many requests all missing on the same cold key at once), look into “cache stampede” protection, typically a per-key lock so only one request repopulates a given key while the rest wait for that result instead of all hitting the expensive path simultaneously.








No Comment! Be the first one.