Fading Coder

One Final Commit for the Last Sprint

Home > Tech > Content

Practical Caching Techniques for Python Applications to Boost Runtime Performance

Tech 1

Using functools.lru_cache for Out-of-the-Box Caching

The functools standard library includes the lru_cache decorator, wich adds least-recently-used caching to any callable with zero extra setup.

from functools import lru_cache

@lru_cache(maxsize=256)
def calc_fib_sequence(pos):
    if pos < 2:
        return pos
    return calc_fib_sequence(pos - 1) + calc_fib_sequence(pos - 2)

# First execution: no cache hit, full computation runs
print(calc_fib_sequence(25))
# Second execution: returns cached result instantly
print(calc_fib_sequence(25))

Build a Custom Cache Decorator

For use cases that require custom logic not covered by lru_cache, you can implement a lightweight in-memory cache decorator from scratch.

def in_memory_cache(func):
    stored_results = {}

    def cache_handler(*args, **kwargs):
        # Use combined args and sorted kwargs as cache key to support keyword parameters
        cache_key = (args, tuple(sorted(kwargs.items())))
        if cache_key in stored_results:
            return stored_results[cache_key]
        output = func(*args, **kwargs)
        stored_results[cache_key] = output
        return output

    return cache_handler

@in_memory_cache
def get_fib_value(n):
    if n < 2:
        return n
    return get_fib_value(n-1) + get_fib_value(n-2)

# First call computes and stores result
print(get_fib_value(25))
# Second call fetches from stored results
print(get_fib_value(25))

Common Caching Use Cases

Caching delivers the most value for I/O bound or compute-heavy operations that repeat with identical inputs. Typical use cases include:

  • Results of repeated relational database or NoSQL queries
  • Responses from external API calls with static payloads
  • Output of long-running mathematical computations or data transformation pipelines
@lru_cache(maxsize=512)
def fetch_db_records(sql_query):
    # Simulate slow database I/O operation
    # Actual implementation would connect to database and execute query
    mock_result = [{"id": i, "value": f"record_{i}"} for i in range(10)]
    return mock_result

# First call triggers database connection and query execution
first_fetch = fetch_db_records("SELECT id, value FROM product_stock")
# Second identical call returns cached result without database access
second_fetch = fetch_db_records("SELECT id, value FROM product_stock")

Cache Expiration and Invalidation

Stale cached data can lead to incorrect application behavior, so you need to implement proper invalidation rules.

Time-Based Expiration

Attach a time-to-live (TTL) value to cached entries, and purge entries that exceed the TTL threshold on access.

from functools import lru_cache
import time

@lru_cache(maxsize=256)
def time_bound_cached_call(input_val):
    TTL_SECONDS = 120
    # Initialize timestamp attribute on first function run
    if not hasattr(time_bound_cached_call, "last_purge_time"):
        time_bound_cached_call.last_purge_time = time.time()
    
    current_timestamp = time.time()
    if current_timestamp - time_bound_cached_call.last_purge_time > TTL_SECONDS:
        time_bound_cached_call.cache_clear()
        time_bound_cached_call.last_purge_time = current_timestamp
    
    # Replace with actual business logic
    return input_val * 10

Scheduled Periodic Purge

For high-throughput applications, run a background task to clear cached data on a fixed enterval instead of checking TTL on every access.

from threading import Timer
from functools import lru_cache

@lru_cache(maxsize=256)
def periodic_cached_operation(param):
    # Business logic implementation
    return f"processed_{param}"

def scheduled_cache_purge():
    periodic_cached_operation.cache_clear()
    # Schedule next purge run
    Timer(120, scheduled_cache_purge).start()

# Start background purge task on application launch
Timer(120, scheduled_cache_purge).start()

Invalidation Safety Notes

Always account for concurrent access when implementing cache modification logic, use locks for shared cache instances to avoid race conditions. Balance TTL length and purge frequenyc to avoid excess overhead from frequent cache clearing.

Custom Cache Eviction Policies

lru_cache uses the least recently used eviction policy by default, but you can implement custom policies to match your access patterns:

  • LFU (Least Frequently Used): Evicts entries that are accessed the least number of times, ideal for workloads with stable frequent access patterns
  • FIFO (First In First Out): Evicts the oldest entry regardless of access frequency, suitable for simple workloads with low cache overhead
  • LRU-K: Evicts entries based on the K-th most recent access, balances recency and frequency to reduce cache misses for bursty workloads
from collections import OrderedDict

class EvictingCache:
    def __init__(self, max_capacity):
        self.max_capacity = max_capacity
        self.cache_store = OrderedDict()

    def __call__(self, func):
        def execute_with_cache(*args):
            if args in self.cache_store:
                # Move accessed entry to end to mark as recently used
                self.cache_store.move_to_end(args)
                return self.cache_store[args]
            output = func(*args)
            if len(self.cache_store) >= self.max_capacity:
                # Remove oldest entry from front of OrderedDict
                self.cache_store.popitem(last=False)
            self.cache_store[args] = output
            return output
        return execute_with_cache

@EvictingCache(max_capacity=128)
def compute_square_value(input_num):
    return input_num ** 2

Caching Performance Best Practices

Memory Footprint and Capacity Limits

Set appropriate maximum cache sizes based on the size of individual cached entries. Oversized caches can lead to high memory usage and degrade overall system performance.

Stale Data Management

Implement regular purge routines for expired entries to avoid holding unused data in memory for extended periods.

Cache Hit Rate Monitoring

Track cache hit rates to evaluate the effectiveness of your caching strategy. A hit rate below 70% usually indicates a misaligned cache policy or changing access patterns that require adjustment.

Concurrent Access Safety

Add thread locks or use process-safe cache implementations for multi-threaded or multi-process applications to prevent data inconsistency and race conditions.

Cold Start Mitigation

Preload frequently accessed data into the cache during application startup to avoid performance dips from initial cache misses after deployment or cache purge.

TTL Configuration

Set explicit TTL values for all cached entries, especially for data that changes regularly, to prevent serving stale data to end users.

Related Articles

Understanding Strong and Weak References in Java

Strong References Strong reference are the most prevalent type of object referencing in Java. When an object has a strong reference pointing to it, the garbage collector will not reclaim its memory. F...

Comprehensive Guide to SSTI Explained with Payload Bypass Techniques

Introduction Server-Side Template Injection (SSTI) is a vulnerability in web applications where user input is improper handled within the template engine and executed on the server. This exploit can r...

Implement Image Upload Functionality for Django Integrated TinyMCE Editor

Django’s Admin panel is highly user-friendly, and pairing it with TinyMCE, an effective rich text editor, simplifies content management significantly. Combining the two is particular useful for bloggi...

Leave a Comment

Anonymous

◎Feel free to join the discussion and share your thoughts.