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.

PythonMini example
def d(f): return lambda x: f(x) * 2
@d
def g(x): return x + 1
print(g(3))
8
Watch out

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.

PythonMini example
from collections import OrderedDict; d = OrderedDict({'a': 1, 'b': 2}); d.move_to_end('a'); print(list(d.keys()))
['b', 'a']
Watch out

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.

PythonMini example
entry = (100, 15.0); now = 20.0; print(now > entry[1])
True
Watch out

Relying on scheduled background cleanup threads introduces lock contention compared to checking expiry lazily on read.

Architecture and data flow

Problem scenario diagram for py-proj-lru-cache-decorator
Problem scenario diagram for py-proj-lru-cache-decorator. Figure 1: Operational problem and business context
Data flow diagram for py-proj-lru-cache-decorator
Data flow diagram for py-proj-lru-cache-decorator. Figure 2: Pipeline execution and data transformation flow
Result shape diagram for py-proj-lru-cache-decorator
Result shape diagram for py-proj-lru-cache-decorator. Figure 3: Output verification and method results

The data

Catalog query lookup inputs with product SKU and currency code.

catalog_queries: Repeated price query access stream

skucurrencyquery_type
SKU-100INRprice_lookup
SKU-100INRprice_lookup
SKU-200USDprice_lookup
SKU-100INRprice_lookup

Step by step

Read each step, run its code, then check your output against the one shown.

  1. 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.

    PythonStep 1
    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 mistake

    Watch out: Using unhashable types like lists or dicts directly inside *args causes TypeError: unhashable type when used as dictionary keys.

    How you will know

    The expensive lookup message prints only once, confirming that the second call returns the cached result.

  2. 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.

    PythonStep 2
    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 mistake

    Watch out: Unbounded caches consume all available process RAM over time if high-cardinality keys are not bounded by a capacity limit.

    How you will know

    The output shows 1,000 entries accumulated indefinitely, illustrating unbounded growth.

  3. 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.

    PythonStep 3
    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 mistake

    Watch out: Omitting move_to_end() on cache hits causes frequently accessed items to be evicted prematurely as if they were cold.

    How you will know

    Key B is evicted because A was touched before D was inserted.

  4. 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.

    PythonStep 4
    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 mistake

    Watch out: Forgetting to delete expired keys from the dictionary leaves stale records taking up cache capacity slots.

    How you will know

    Lookup at t=5.0s returns 83.5, while lookup at t=15.0s returns None and evicts the key from cache.

  5. 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.

    PythonStep 5
    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 mistake

    Watch out: Omitting the nonlocal keyword when incrementing hits and misses causes UnboundLocalError in nested wrapper scopes.

    How you will know

    Telemetry 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.

Python 3Ready to run
Loading editor…
Press Run Code to evaluate
LRU TTL decorator executes cleanly
Returns accurate price values
Accurately records 1 hit and 2 misses

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

OutputCopy this line
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

OutputREADME.md
# 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.
Stretch goal

Support threading locks (threading.RLock) to ensure thread-safe cache mutations across multi-threaded web application workers.

Keep going