Introduction

Have you ever written a function that re-computes the same expensive result over and over? Enter memoization — caching function results so repeated calls with the same arguments return instantly. Python's functools.lru_cache provides a simple, well-tested way to memoize pure functions.

In this post we'll:

• Explain the core concepts and prerequisites.
• Walk through hands-on, real-world examples.
• Show how lru_cache interacts with parallelism tools such as concurrent.futures and multiprocessing.
• Demonstrate patterns for using lru_cache in interactive dashboards (Dash).
• Share best practices, pitfalls, and advanced tips.

Prerequisites

• Python 3.7+ recommended (behaviour consistent across 3.7–3.12).
• Familiarity with functions, decorators, and basic concurrency (threads/processes).
• Optional: basic exposure to Dash and web frameworks if you plan to integrate caching in an app.

• Official docs: https://docs.python.org/3/library/functools.html#functools.lru_cache

Core Concepts

• Memoization: Storing results of expensive function calls and returning the cached result when inputs repeat.
• LRU (Least Recently Used): lru_cache evicts the least recently used entry when the cache is full.
• Cache key: Based on function arguments; arguments must be hashable (immutable types).
• Thread-safety: lru_cache is thread-safe (it uses an internal lock) in CPython.
• Process-safety: lru_cache is not shared between processes; each process has its own cache.

Basic Example: Fibonacci with lru_cache

A classic demonstration: recursive Fibonacci without memoization is exponential time.

from functools import lru_cache
@lru_cache(maxsize=128)
def fib(n: int) -> int:
    """Return the nth Fibonacci number (naive recursive with memoization)."""
    if n < 2:
        return n
    return fib(n - 1) + fib(n - 2)

Explanation (line-by-line):

1. from functools import lru_cache — import the decorator.
2. @lru_cache(maxsize=128) — wrap fib with an LRU cache holding up to 128 results.
3. def fib(n: int) -> int: — define a typed function for clarity.
4. Base case: if n < 2, return n.
5. Recursive case: return sum of previous two Fibonacci numbers; repeated subcalls are cached.

• Non-integer or negative n should be validated in production code.
• maxsize=None would create an unbounded cache (use carefully).

• fib(35) is instantaneous after initial recursion completes; subsequent calls are cached.

print(fib.cache_info())
Output: CacheInfo(hits=..., misses=..., maxsize=128, currsize=...)

Real-World Example: Caching Expensive I/O or Computations

Imagine a function that queries a slow external API or performs a heavy computation.

Example: simulated heavy computation (e.g., image processing, ML inference):

import time
from functools import lru_cache
@lru_cache(maxsize=256)
def heavy_compute(key: str) -> dict:
    """Simulate a heavy computation or API call returning JSON-like data."""
    # Simulate latency
    time.sleep(2)
    # Example result
    return {"key": key, "value": hash(key) % 1000}

Explanation:

• Each unique key costs ~2 seconds once; further calls return immediately from cache.
• maxsize=256 constrains memory use.

import time
start = time.perf_counter()
print(heavy_compute("alpha"))  # ~2s
print("Elapsed:", time.perf_counter() - start)
start = time.perf_counter()
print(heavy_compute("alpha"))  # ~0s (cached)
print("Elapsed:", time.perf_counter() - start)

Edge cases:

• If underlying data changes (the API result for "alpha" changes), cache will serve stale results. Consider cache invalidation strategies (timeouts, versioned keys).

Measuring Impact: Profiling Cache Effectiveness

Use cache_info() to measure hits/misses.

print(heavy_compute.cache_info())
Example output: CacheInfo(hits=1, misses=1, maxsize=256, currsize=1)

High hit ratio indicates effective memoization.

Integrating lru_cache with concurrent.futures

• With ThreadPoolExecutor, threads share the same memory space; cache is shared across threads and thread-safe.
• With ProcessPoolExecutor or multiprocessing, each process has its own memory; each process has its own cache, so benefits are per-process only.

from concurrent.futures import ThreadPoolExecutor
import time
@lru_cache(maxsize=128)
def fib_cached(n: int) -> int:
    if n < 2:
        return n
    return fib_cached(n - 1) + fib_cached(n - 2)
def call_fib(n):
    return fib_cached(n)
if __name__ == "__main__":
    with ThreadPoolExecutor(max_workers=4) as ex:
        results = list(ex.map(call_fib, [30, 31, 32, 30, 31]))
    print("Cache info:", fib_cached.cache_info())

Explanation:

• Threads share the same process memory and the same fib_cached cache.
• Concurrent calls can benefit from previously computed results.

from concurrent.futures import ProcessPoolExecutor
def call_fib_in_process(n):
    # Each worker process will import and create its own lru_cache for fib_cached
    from functools import lru_cache
    @lru_cache(maxsize=128)
    def local_fib(n):
        if n < 2:
            return n
        return local_fib(n - 1) + local_fib(n - 2)
    return local_fib(n)
if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as ex:
        results = list(ex.map(call_fib_in_process, [30, 31, 32, 30, 31]))
    print(results)

Explanation:

• Each process builds its own cache; repeated values across tasks may be recomputed in each worker, losing the benefit of a shared cache.

• Use ThreadPool when your task is I/O-bound and you want a shared cache.
• Use ProcessPool for CPU-bound tasks (to bypass GIL), but be mindful that lru_cache will be per-process.

Sharing Cached Results Across Processes

If you need a cache shared across processes, consider:

• A multiprocessing.Manager().dict() as a shared cache.
• Using an external cache like Redis, Memcached, or in-process web server caches (e.g., Flask-Caching).
• Using file-based caches or sqlite.

from multiprocessing import Manager, Pool
import time
import hashlib
import json
def init(shared_cache):
    global cache
    cache = shared_cache
def make_key(args):
    # Deterministic key for arguments - use careful hashing in real cases
    return hashlib.sha256(json.dumps(args, sort_keys=True).encode()).hexdigest()
def compute(data):
    k = make_key(data)
    if k in cache:
        return cache[k]  # shared across processes
    # Simulate heavy work
    time.sleep(2)
    result = {"result": sum(data)}  # placeholder
    cache[k] = result
    return result
if __name__ == "__main__":
    manager = Manager()
    shared_cache = manager.dict()
    with Pool(initializer=init, initargs=(shared_cache,)) as p:
        inputs = [[1, 2, 3], [4, 5, 6], [1, 2, 3]]
        results = p.map(compute, inputs)
    print(results)

Notes:

• Manager-based caches add IPC overhead; good for moderate sharing, not high-performance low-latency caching.
• In production, use Redis or a dedicated caching layer if you need high throughput and persistence.

Using lru_cache in a Dash App

Dash apps often benefit from caching computed results for interactive callbacks. If your Dash app runs in a single process (development or a single worker), lru_cache can be a quick win for caching pure callbacks.

Example (conceptual snippet):

# app.py
from functools import lru_cache
import dash
from dash import html, dcc
import time
app = dash.Dash(__name__)
@lru_cache(maxsize=128)
def expensive_query(param: str):
    time.sleep(2)  # simulate expensive work
    return f"Result for {param}"
app.layout = html.Div([
    dcc.Input(id="input", value="alpha"),
    html.Button("Compute", id="button"),
    html.Div(id="output")
])
@app.callback(
    dash.dependencies.Output("output", "children"),
    [dash.dependencies.Input("button", "n_clicks")],
    [dash.dependencies.State("input", "value")]
)
def on_click(n_clicks, value):
    if not n_clicks:
        return "Press compute"
    return expensive_query(value)
if __name__ == "__main__":
    app.run_server(debug=True)

Important considerations:

• For multi-worker deployments (Gunicorn/WSGI with multiple processes), each worker maintains its own lru_cache. To have a centralized cache, use flask_caching with Redis or filesystem backend.
• For async callbacks or async frameworks, lru_cache is synchronous. For async caching consider aiocache or server-side caches.

Advanced: Custom Keys, Unhashable Arguments, and Typed Option

• lru_cache requires function arguments to be hashable. Mutable types (lists, dicts) will raise TypeError.
• You can convert mutable inputs to immutable representations (tuples, frozensets, or serialized JSON) for caching keys.

from functools import lru_cache, wraps
import json
def hashable_args(func):
    @wraps(func)
    def wrapper(args, kwargs):
        # Convert args/kwargs to string representation; ensure deterministic ordering
        key = json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True, default=str)
        return func(key)
    return wrapper

@lru_cache(maxsize=128)
@hashable_args
def compute_from_key(key_str):
    # Since compute_from_key receives a key string, it can reconstruct
    # or parse key_str if needed, or simply use the key to store results.
    return f"Computed {key_str}"

1. Unhashable Arguments: Convert to immutable types or compute a stable key.
2. Stale Data: Cache doesn't know when the underlying data changes. Strategies: time-based invalidation, versioned keys, or explicit cache_clear().
3. Memory Growth: Unbounded caches (maxsize=None) can cause OOM. Always consider a reasonable maxsize.
4. Side Effects: Functions with side effects (writing to DB, network calls) are poor candidates for memoization.
5. Exception Handling: If a function raises an exception, the result isn't cached. You may want to catch and handle expected exceptions inside the function.
6. Multiprocessing Misunderstanding: Expecting a single shared cache across processes leads to surprises — consider external caches.

from concurrent.futures import ProcessPoolExecutor, as_completed
from functools import lru_cache
@lru_cache(maxsize=1024)
def preprocess_key(k):
    # cheap canonicalization, shared in main process
    return k.strip().lower()
def heavy_task(k):
    # executed in worker processes, no shared lru_cache here
    import time
    time.sleep(2)
    return k, len(k)
def batch_process(keys):
    unique = {}
    # Preprocess and use main-process cache
    for k in keys:
        pk = preprocess_key(k)
        unique.setdefault(pk, []).append(k)
    results = {}
    with ProcessPoolExecutor() as ex:
        futures = {ex.submit(heavy_task, pk): pk for pk in unique}
        for fut in as_completed(futures):
            pk = futures[fut]
            results[pk] = fut.result()
    return results

import time
from functools import lru_cache
def with_ttl(ttl_seconds):
    def decorator(func):
        @lru_cache(maxsize=1024)
        def wrapper(args, _ts, *kwargs):
            return func(args, *kwargs)
        def wrapped(args, *kwargs):
            # compute bucketed timestamp to invalidate cache periodically
            ts = int(time.time() / ttl_seconds)
            return wrapper(args, _ts=ts, *kwargs)
        wrapped.cache_clear = wrapper.cache_clear
        wrapped.cache_info = wrapper.cache_info
        return wrapped
    return decorator

@with_ttl(60)
def data_fetch(x):
    # expensive fetch
    return x  2

Note: This creates additional synthetic cache key _ts to expire results every ttl_seconds.

Conclusion

• It's per-process: use external caches for multi-process sharing.
• Arguments must be hashable or normalized.
• Consider memory limits and invalidation needs.

Call to Action

Try the examples in this post:

• Experiment with cache_info() and adjust maxsize.
• Convert a real slow function in your project to use lru_cache and measure speedups.
• If you deploy a Dash app, try lru_cache for local caches and evaluate a Redis backend for multi-worker deployments.

Further Reading

• Official functools docs: https://docs.python.org/3/library/functools.html#functools.lru_cache
• concurrent.futures docs: https://docs.python.org/3/library/concurrent.futures.html
• multiprocessing docs: https://docs.python.org/3/library/multiprocessing.html
• Dash docs & caching: https://dash.plotly.com/ and https://flask-caching.readthedocs.io/
• cachetools library: https://cachetools.readthedocs.io/