pythonintermediate45 minPython 3
Custom LRU Cache with TTL Expiration
Build a custom Least Recently Used (LRU) function decorator in Python with time-to-live (TTL) expiration and capacity eviction.
The incident
- Company
- OmniCart Tech
- Engineer
- Rohan Kulkarni, Platform Performance Engineer
- Trigger
- Database connection pool exhaustion caused by identical catalog price and currency exchange queries executed millions of times per hour.
- Stakes
- Over 4,200 identical catalog queries per second overwhelmed the replica database, spiking query latency from 8ms to 650ms and risking site-wide checkout slowdowns.
During peak afternoon shopping hours across OmniCart Tech e-commerce platforms, database monitoring dashboards flashed critical red alerts across engineering operations centers. The PostgreSQL read-replica cluster handling product catalog queries was operating at ninety-eight percent CPU capacity, with available database connection pools hovering near total exhaustion. Product page rendering times across mobile apps degraded sharply from an eight-millisecond baseline to over six hundred and fifty milliseconds, creating severe friction across customer shopping funnels, increasing bounce rates, and stalling high-value purchases across multiple product categories.
The database infrastructure team performed an emergency query analysis and discovered that over forty-two hundred queries per second were requesting identical, immutable product pricing and currency conversion rates. Frontend recommendation engines, personalized banners, and category search filters were hitting the database redundantly on every page view rather than reusing recently computed pricing calculations in local worker memory, saturating connection sockets with repeated lookups for the exact same catalog records and overwhelming the storage subsystem with unneeded read I/O operations.
The commercial risk was immediate: site performance audits showed that every hundred milliseconds of checkout latency reduced mobile shopping cart conversion rates by two percent. If the database connection pool completely saturated, the entire product catalog would throw HTTP 503 service unavailable errors, risking estimated losses exceeding twenty-five lakh rupees during high-margin promotional campaigns across major regional markets, while customer dissatisfaction escalated rapidly on social media and partner merchant portals across regional retail sectors.
Rohan Kulkarni, a platform performance engineer on the core site reliability team, was tasked with designing a lightweight, in-memory caching decorator in Python that could wrap expensive backend database fetch functions without modifying existing application business logic. While Python standard library includes functools.lru_cache, it lacks built-in Time-To-Live (TTL) expiration, which was mandatory because currency conversion rates and promotional discount prices updated every fifteen minutes across regional marketing campaigns and partner merchant channels, requiring periodic invalidation.
Rohan structured the solution around collections.OrderedDict to deliver constant O(1) lookup and eviction performance while enforcing strict maximum memory ceilings across container pods. Each cached entry stored the evaluated return value alongside an expiration timestamp. When a cached function was called, the decorator verified whether the cached item had expired; if stale, it evicted the entry and re-executed the underlying function to fetch fresh database values cleanly without leaking memory or retaining outdated pricing metrics across long-running application threads.
Deploying the custom LRU-TTL caching decorator to production microservices dropped read-replica database CPU utilization from ninety-eight percent to under twelve percent within minutes. Repeated catalog queries completed in microseconds directly from memory, while TTL boundaries ensured that promotional price changes propagated accurately across shopping carts. The caching framework results are reviewed weekly by the Principal Systems Architect and Backend Engineering Lead to guarantee stable catalog performance during high-concurrency seasonal sales.
What you will learn
- Implement Python function decorators using functools.wraps
- Maintain Least Recently Used (LRU) eviction orders using collections.OrderedDict
- Enforce Time-To-Live (TTL) cache expiration on timestamped entries
- Track cache hit and miss telemetry for performance optimization
What you need first
Python DecoratorsPython functions and decorators
Decorators wrap callable functions to intercept arguments, modify execution, and manage return values transparently using functools.wraps.
def d(f): return lambda x: f(x) * 2
@d
def g(x): return x + 1
print(g(3))8
Forgetting @functools.wraps(fn) obscures original function docstrings and reflection metadata (__name__).
OrderedDict for LRU OrderingPython hash maps
The collections.OrderedDict class maintains key insertion order and supports O(1) move_to_end() and popitem(last=False) operations.
from collections import OrderedDict; d = OrderedDict({'a': 1, 'b': 2}); d.move_to_end('a'); print(list(d.keys()))['b', 'a']
Using standard Python dict without move_to_end requires slow O(n) re-insertions to maintain recency ordering.
TTL Cache ExpirationDates and timestamps in Python
Storing timestamps alongside cached payloads allows evaluating now > expiry to invalidate stale data lazily upon lookup.
entry = (100, 15.0); now = 20.0; print(now > entry[1])True
Relying on scheduled background cleanup threads introduces lock contention compared to checking expiry lazily on read.
Architecture and data flow
The data
Catalog query lookup inputs with product SKU and currency code.
catalog_queries: Repeated price query access stream
| sku | currency | query_type |
|---|---|---|
| SKU-100 | INR | price_lookup |
| SKU-100 | INR | price_lookup |
| SKU-200 | USD | price_lookup |
| SKU-100 | INR | price_lookup |
Step by step
Read each step, run its code, then check your output against the one shown.
Define basic function caching decorator pattern
We construct the basic structure of a Python function decorator using functools.wraps. The outer function accepts configuration parameters, while the inner wrapper intercepts calls to the target function, generating a composite cache key from the supplied positional and keyword arguments. In high-performance backend systems, memoization prevents redundant executions of expensive database queries or mathematical calculations. We confirm that the wrapper invokes the target function and stores results in a local dictionary for rapid reuse. This caching mechanism guarantees constant time overhead.
from functools import wraps def simple_cache(fn): cache = {} @wraps(fn) def wrapper(*args): if args not in cache: cache[args] = fn(*args) return cache[args] return wrapper @simple_cache def fetch_rate(base: str, target: str) -> float: print(f"Executing expensive lookup: {base}->{target}") return 83.5 r1 = fetch_rate("USD", "INR") r2 = fetch_rate("USD", "INR") print(f"Results: {r1}, {r2}")Executing expensive lookup: USD->INR Results: 83.5, 83.5
Common mistakeWatch out: Using unhashable types like lists or dicts directly inside *args causes TypeError: unhashable type when used as dictionary keys.
How you will knowThe expensive lookup message prints only once, confirming that the second call returns the cached result.
Demonstrate pitfalls of unbounded dictionary memory growth
Wrong turn first: this step shows the mistake before the fix
We demonstrate why unbounded dictionary caches cause out-of-memory crashes in production microservices. When a naive caching decorator stores every unique query parameter indefinitely without capacity limits, high-cardinality queries rapidly accumulate millions of dictionary entries. As simulated by our benchmark loop, memory use grows with the number of distinct keys, eventually triggering OS kill signals (SIGKILL) on container hosts during traffic surges. Bounding capacity is mandatory for reliable service operations and predictable memory usage. This caching mechanism guarantees constant time overhead.
unbounded_cache = {} # Simulate high-cardinality query keys flooding the cache for i in range(1000): unbounded_cache[f"user_session_{i}"] = "session_token_data" print(f"Unbounded cache entries: {len(unbounded_cache)} (Memory leaks without eviction limits!)")Unbounded cache entries: 1000 (Memory leaks without eviction limits!)
Common mistakeWatch out: Unbounded caches consume all available process RAM over time if high-cardinality keys are not bounded by a capacity limit.
How you will knowThe output shows 1,000 entries accumulated indefinitely, illustrating unbounded growth.
Implement Least Recently Used (LRU) capacity eviction
To enforce strict memory boundaries, we use collections.OrderedDict to manage cache capacity. When a key is accessed, we call cache.move_to_end(key) to mark it as the most recently used entry. When the cache reaches its maximum capacity and a new key must be inserted, we invoke cache.popitem(last=False) to evict the oldest, least recently accessed entry in O(1) time. We confirm that adding a fourth key to a capacity-3 cache cleanly evicts the least recently accessed item from memory. This caching mechanism guarantees constant time overhead.
from collections import OrderedDict cache = OrderedDict() max_size = 3 def set_key(k: str, v: str) -> str: evicted = None if k in cache: cache.move_to_end(k) else: if len(cache) >= max_size: evicted, _ = cache.popitem(last=False) cache[k] = v return evicted set_key("A", "data_A") set_key("B", "data_B") set_key("C", "data_C") # Touch A to make it recently used set_key("A", "data_A") # Insert D -> should evict B (since A was refreshed) evicted = set_key("D", "data_D") print(f"Evicted key: {evicted} | Remaining keys: {list(cache.keys())}")Evicted key: B | Remaining keys: ['C', 'A', 'D']
Common mistakeWatch out: Omitting move_to_end() on cache hits causes frequently accessed items to be evicted prematurely as if they were cold.
How you will knowKey B is evicted because A was touched before D was inserted.
Incorporate Time-To-Live (TTL) expiration timestamps
Financial conversion rates and promotional discounts fluctuate over time, making permanent caching unsafe. We augment each cached payload with an expiration timestamp: (value, expiry_timestamp). On every lookup, the decorator compares the current timestamp against the stored expiration time. If now exceeds expiry, the entry is invalidated, evicted from the dictionary, and recomputed fresh from the underlying data source. We verify that expired entries trigger a re-fetch rather than serving stale metrics to downstream consumers. This caching mechanism guarantees constant time overhead.
from collections import OrderedDict cache = OrderedDict() ttl_seconds = 10.0 def get_with_ttl(key: str, now: float) -> str: if key in cache: val, expiry = cache[key] if now < expiry: cache.move_to_end(key) return val # Expired: evict del cache[key] return None # Seed key at t=0 cache["USD_INR"] = (83.5, 0.0 + ttl_seconds) print(f"Lookup at t=5.0s: {get_with_ttl('USD_INR', now=5.0)}") print(f"Lookup at t=15.0s: {get_with_ttl('USD_INR', now=15.0)} (Expired!)") print(f"Cache keys after expiry: {list(cache.keys())}")Lookup at t=5.0s: 83.5 Lookup at t=15.0s: None (Expired!) Cache keys after expiry: []
Common mistakeWatch out: Forgetting to delete expired keys from the dictionary leaves stale records taking up cache capacity slots.
How you will knowLookup at t=5.0s returns 83.5, while lookup at t=15.0s returns None and evicts the key from cache.
Build comprehensive lru_ttl_cache decorator with hit/miss metrics
In our final milestone, we unify capacity bounding, LRU eviction, and TTL expiration into a reusable decorator named lru_ttl_cache. We also attach an internal metrics counter to record cache hits and misses. We wrap an expensive simulated database query function, execute a multi-call workload with both repeats and expirations, and print the resulting telemetry summary to confirm optimal operational performance and memory management across production instances. This caching mechanism guarantees constant time overhead.
from collections import OrderedDict from functools import wraps def lru_ttl_cache(max_size: int = 3, ttl: float = 10.0): def decorator(fn): cache = OrderedDict() hits = 0 misses = 0 @wraps(fn) def wrapper(*args, now: float = 0.0): nonlocal hits, misses if args in cache: val, expiry = cache[args] if now < expiry: cache.move_to_end(args) hits += 1 return val del cache[args] misses += 1 result = fn(*args) if len(cache) >= max_size: cache.popitem(last=False) cache[args] = (result, now + ttl) return result wrapper.metrics = lambda: {"hits": hits, "misses": misses} return wrapper return decorator @lru_ttl_cache(max_size=2, ttl=5.0) def get_price(sku: str) -> float: return 199.0 if sku == "SKU-1" else 499.0 # 1. First call: miss get_price("SKU-1", now=0.0) # 2. Immediate second call: hit get_price("SKU-1", now=1.0) # 3. Call after TTL expired: miss get_price("SKU-1", now=6.0) print(f"Cache telemetry: {get_price.metrics()}")Cache telemetry: {'hits': 1, 'misses': 2}Common mistakeWatch out: Omitting the nonlocal keyword when incrementing hits and misses causes UnboundLocalError in nested wrapper scopes.
How you will knowTelemetry confirms exactly {'hits': 1, 'misses': 2} across the 3 calls.
Build it yourself
Write your solution below, run it in your browser, and pass every check.
What the result means
Interpretation
The custom LRU-TTL caching decorator prevents redundant database queries by serving repeat requests directly from memory. In production, this architecture drops database connection pool utilization from near exhaustion back to healthy baselines.
The message you would send
Hi Platform Performance Team, the custom LRU-TTL cache decorator has been validated. The new utility combines O(1) capacity bounding (OrderedDict) with lazy TTL expiration, delivering sub-millisecond cached responses.
Put it on your CV
Resume or LinkedIn bullet
Engineered a high-performance LRU cache decorator in Python with TTL expiration and OrderedDict capacity bounding, reducing database CPU load by 86% across catalog microservices.README for GitHub
# Custom LRU Cache with TTL Expiration
## Business Context
Redundant catalog price and currency conversion database queries overwhelmed database replicas during peak traffic, spiking query latency from 8ms to 650ms.
## Solution Architecture
- Implemented a Python function decorator with `@functools.wraps` metadata preservation.
- Utilized `collections.OrderedDict` to provide constant O(1) LRU eviction order updates.
- Integrated lazy timestamp-based Time-To-Live (TTL) expiration to purge stale exchange rates.
## Verification
Passes all automated verification checks on Python 3 standard library.Support threading locks (threading.RLock) to ensure thread-safe cache mutations across multi-threaded web application workers.