Python functools.lru_cache and cache: Supercharge Performance with Memoization
Performance optimization is crucial in Python applications, and one of the most effective techniques is memoization - storing the results of expensive function calls and returning cached results when the same inputs occur again. Python's functools
module provides powerful tools for this: lru_cache
and the newer cache
decorator. This comprehensive guide will show you how to use these tools to dramatically improve your application's performance.
Table Of Contents
- Understanding Memoization and Caching
- functools.lru_cache: Least Recently Used Caching
- functools.cache: Simplified Unlimited Caching
- Real-World Applications
- Advanced Caching Patterns
- Performance Monitoring and Optimization
- Best Practices and Guidelines
- Conclusion
Understanding Memoization and Caching
Memoization is an optimization technique where you cache the results of function calls. When the function is called again with the same arguments, instead of recomputing the result, you return the cached value. This can provide massive performance improvements for functions with expensive computations or repetitive calls.
from functools import lru_cache
import time
# Without caching - slow recursive fibonacci
def fibonacci_slow(n):
if n < 2:
return n
return fibonacci_slow(n-1) + fibonacci_slow(n-2)
# With caching - lightning fast
@lru_cache(maxsize=None)
def fibonacci_fast(n):
if n < 2:
return n
return fibonacci_fast(n-1) + fibonacci_fast(n-2)
# Performance comparison
start = time.time()
result_slow = fibonacci_slow(30)
slow_time = time.time() - start
start = time.time()
result_fast = fibonacci_fast(30)
fast_time = time.time() - start
print(f"Slow version: {result_slow} in {slow_time:.4f} seconds")
print(f"Fast version: {result_fast} in {fast_time:.6f} seconds")
print(f"Speedup: {slow_time/fast_time:.0f}x faster!")
functools.lru_cache: Least Recently Used Caching
lru_cache
implements a Least Recently Used caching strategy, where the least recently accessed items are discarded when the cache reaches its maximum size.
Basic Usage
from functools import lru_cache
@lru_cache(maxsize=128)
def expensive_computation(x, y):
"""Simulate an expensive computation."""
print(f"Computing {x} + {y}...") # This will only print on cache misses
time.sleep(0.1) # Simulate work
return x + y
# First calls - cache misses
print(expensive_computation(1, 2)) # Prints "Computing 1 + 2..."
print(expensive_computation(3, 4)) # Prints "Computing 3 + 4..."
# Subsequent calls - cache hits (much faster)
print(expensive_computation(1, 2)) # No print, returns cached result
print(expensive_computation(3, 4)) # No print, returns cached result
# Cache statistics
print(expensive_computation.cache_info())
# CacheInfo(hits=2, misses=2, maxsize=128, currsize=2)
maxsize Parameter
The maxsize
parameter controls cache size and behavior:
from functools import lru_cache
# Limited cache size
@lru_cache(maxsize=2)
def limited_cache(n):
print(f"Processing {n}")
return n * n
# Unlimited cache size
@lru_cache(maxsize=None)
def unlimited_cache(n):
print(f"Processing {n}")
return n * n
# Test limited cache
for i in [1, 2, 3, 1, 2, 3]: # Will evict items due to size limit
limited_cache(i)
print("Limited cache info:", limited_cache.cache_info())
# Test unlimited cache
for i in [1, 2, 3, 1, 2, 3]: # All items stay cached
unlimited_cache(i)
print("Unlimited cache info:", unlimited_cache.cache_info())
functools.cache: Simplified Unlimited Caching
Python 3.9 introduced functools.cache
, which is equivalent to lru_cache(maxsize=None)
but with better performance:
from functools import cache
import time
@cache
def prime_factors(n):
"""Find prime factors of a number."""
print(f"Calculating prime factors of {n}")
factors = []
d = 2
while d * d <= n:
while n % d == 0:
factors.append(d)
n //= d
d += 1
if n > 1:
factors.append(n)
return factors
# Usage
print(prime_factors(60)) # Cache miss - calculation performed
print(prime_factors(84)) # Cache miss - calculation performed
print(prime_factors(60)) # Cache hit - instant return
print("Cache info:", prime_factors.cache_info())
Real-World Applications
Database Query Optimization
Cache expensive database queries:
from functools import lru_cache
import sqlite3
from typing import List, Tuple
class UserService:
def __init__(self, db_path: str):
self.db_path = db_path
@lru_cache(maxsize=100)
def get_user_by_id(self, user_id: int) -> Tuple[str, str]:
"""Get user data with caching."""
print(f"Querying database for user {user_id}")
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("SELECT name, email FROM users WHERE id = ?", (user_id,))
result = cursor.fetchone()
conn.close()
return result or (None, None)
@lru_cache(maxsize=50)
def get_user_posts_count(self, user_id: int) -> int:
"""Get user's post count with caching."""
print(f"Counting posts for user {user_id}")
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM posts WHERE user_id = ?", (user_id,))
count = cursor.fetchone()[0]
conn.close()
return count
def clear_user_cache(self, user_id: int = None):
"""Clear cache for specific user or all users."""
if user_id:
# Clear specific user's cache entries
self.get_user_by_id.cache_clear()
self.get_user_posts_count.cache_clear()
else:
# Clear all caches
self.get_user_by_id.cache_clear()
self.get_user_posts_count.cache_clear()
# Usage
service = UserService("users.db")
user = service.get_user_by_id(1) # Database query
user = service.get_user_by_id(1) # Cached result
API Response Caching
Cache external API responses:
from functools import lru_cache
import requests
import json
from datetime import datetime, timedelta
class WeatherService:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.openweathermap.org/data/2.5"
@lru_cache(maxsize=50)
def get_weather(self, city: str, country: str = "US") -> dict:
"""Get weather data with caching (5-minute cache effective)."""
print(f"Fetching weather for {city}, {country}")
url = f"{self.base_url}/weather"
params = {
"q": f"{city},{country}",
"appid": self.api_key,
"units": "metric"
}
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
# Add timestamp for cache validation
data["cached_at"] = datetime.now().isoformat()
return data
def get_cached_weather(self, city: str, country: str = "US",
max_age_minutes: int = 5) -> dict:
"""Get weather with time-based cache invalidation."""
weather_data = self.get_weather(city, country)
cached_time = datetime.fromisoformat(weather_data["cached_at"])
age = datetime.now() - cached_time
if age > timedelta(minutes=max_age_minutes):
# Clear cache and fetch fresh data
self.get_weather.cache_clear()
weather_data = self.get_weather(city, country)
return weather_data
# Usage
weather = WeatherService("your_api_key")
data = weather.get_cached_weather("London", "UK")
Computational Mathematics
Cache mathematical computations:
from functools import cache
import math
@cache
def factorial(n):
"""Cached factorial calculation."""
if n <= 1:
return 1
return n * factorial(n - 1)
@cache
def binomial_coefficient(n, k):
"""Calculate binomial coefficient with caching."""
if k > n or k < 0:
return 0
if k == 0 or k == n:
return 1
# Use cached factorial
return factorial(n) // (factorial(k) * factorial(n - k))
@cache
def is_prime(n):
"""Check if number is prime with caching."""
if n < 2:
return False
if n == 2:
return True
if n % 2 == 0:
return False
for i in range(3, int(math.sqrt(n)) + 1, 2):
if n % i == 0:
return False
return True
# Usage - calculations are cached for repeated use
print(f"10! = {factorial(10)}")
print(f"C(10,3) = {binomial_coefficient(10, 3)}")
print(f"Is 97 prime? {is_prime(97)}")
# Check cache performance
print("Factorial cache:", factorial.cache_info())
print("Binomial cache:", binomial_coefficient.cache_info())
print("Prime cache:", is_prime.cache_info())
Advanced Caching Patterns
Time-Based Cache Invalidation
Implement time-based cache expiration:
from functools import lru_cache
import time
from datetime import datetime, timedelta
def timed_lru_cache(seconds: int, maxsize: int = 128):
"""Create a cache that expires after specified seconds."""
def decorator_wrapper(func):
cache = lru_cache(maxsize=maxsize)(func)
cache.lifetime = timedelta(seconds=seconds)
cache.expiration = datetime.utcnow() + cache.lifetime
def wrapped_func(*args, **kwargs):
if datetime.utcnow() >= cache.expiration:
cache.cache_clear()
cache.expiration = datetime.utcnow() + cache.lifetime
return cache(*args, **kwargs)
wrapped_func.cache_info = cache.cache_info
wrapped_func.cache_clear = cache.cache_clear
return wrapped_func
return decorator_wrapper
@timed_lru_cache(seconds=10) # Cache expires after 10 seconds
def get_current_time():
"""Get current time - cached for 10 seconds."""
return datetime.now().strftime("%H:%M:%S")
# Test time-based expiration
print(get_current_time()) # Fresh call
time.sleep(2)
print(get_current_time()) # Cached (same time)
time.sleep(9)
print(get_current_time()) # Cache expired, fresh call
Conditional Caching
Cache only under certain conditions:
from functools import lru_cache
def conditional_cache(condition_func):
"""Cache only when condition is met."""
def decorator(func):
cached_func = lru_cache(maxsize=None)(func)
def wrapper(*args, **kwargs):
if condition_func(*args, **kwargs):
return cached_func(*args, **kwargs)
else:
return func(*args, **kwargs)
wrapper.cache_info = cached_func.cache_info
wrapper.cache_clear = cached_func.cache_clear
return wrapper
return decorator
# Only cache for large numbers
def should_cache(n):
return n > 100
@conditional_cache(should_cache)
def expensive_calculation(n):
"""Only cache for n > 100."""
print(f"Calculating for {n}")
return n ** 2 + n * 3 + 1
# Small numbers won't be cached
expensive_calculation(5) # Always calculates
expensive_calculation(5) # Always calculates
# Large numbers will be cached
expensive_calculation(200) # Calculates once
expensive_calculation(200) # Returns cached result
Cache with Custom Key Generation
Control how cache keys are generated:
from functools import lru_cache
import hashlib
import json
class CustomCacheKey:
"""Custom cache key generation for complex objects."""
@staticmethod
def make_key(obj):
"""Create a hashable key from any object."""
if isinstance(obj, dict):
return tuple(sorted(obj.items()))
elif isinstance(obj, list):
return tuple(obj)
elif hasattr(obj, '__dict__'):
return tuple(sorted(obj.__dict__.items()))
else:
return obj
def smart_cache(maxsize=128):
"""Cache with intelligent key generation."""
def decorator(func):
@lru_cache(maxsize=maxsize)
def cached_func(*args, **kwargs):
return func(*args, **kwargs)
def wrapper(*args, **kwargs):
# Convert args and kwargs to hashable types
cache_args = tuple(CustomCacheKey.make_key(arg) for arg in args)
cache_kwargs = tuple(sorted(
(k, CustomCacheKey.make_key(v)) for k, v in kwargs.items()
))
return cached_func(*cache_args, **cache_kwargs)
wrapper.cache_info = cached_func.cache_info
wrapper.cache_clear = cached_func.cache_clear
return wrapper
return decorator
@smart_cache()
def process_config(config_dict, options=None):
"""Process configuration with smart caching."""
print(f"Processing config: {config_dict}")
return f"Processed: {len(config_dict)} settings"
# Works with dictionaries and complex objects
config1 = {"host": "localhost", "port": 8080}
config2 = {"port": 8080, "host": "localhost"} # Same content, different order
process_config(config1)
process_config(config2) # Cache hit due to smart key generation
Performance Monitoring and Optimization
Cache Statistics Analysis
Monitor cache performance:
from functools import lru_cache
import random
@lru_cache(maxsize=100)
def random_computation(seed, iterations):
"""Simulate computation with controlled randomness."""
random.seed(seed)
total = 0
for _ in range(iterations):
total += random.random()
return total
def analyze_cache_performance(func, test_data):
"""Analyze cache hit rate and effectiveness."""
initial_info = func.cache_info()
# Run test
for args in test_data:
func(*args)
final_info = func.cache_info()
hits = final_info.hits - initial_info.hits
misses = final_info.misses - initial_info.misses
total_calls = hits + misses
if total_calls > 0:
hit_rate = hits / total_calls * 100
print(f"Cache Hit Rate: {hit_rate:.1f}%")
print(f"Hits: {hits}, Misses: {misses}")
print(f"Cache Size: {final_info.currsize}/{final_info.maxsize}")
return final_info
# Test data with some repetition
test_data = [
(1, 1000), (2, 500), (1, 1000), # Repeat (1, 1000)
(3, 750), (2, 500), (4, 200), # Repeat (2, 500)
(1, 1000), (5, 300) # Repeat (1, 1000) again
]
analyze_cache_performance(random_computation, test_data)
Best Practices and Guidelines
When to Use Caching
Good candidates for caching:
- Functions with expensive computations
- I/O operations (database, API calls)
- Pure functions (same input → same output)
- Functions called repeatedly with same arguments
from functools import cache
# Good: Expensive mathematical computation
@cache
def matrix_determinant(matrix_tuple):
"""Calculate matrix determinant."""
# Expensive computation here
pass
# Good: External API call
@lru_cache(maxsize=50)
def fetch_user_profile(user_id):
"""Fetch user profile from external API."""
# API call here
pass
Avoid caching for:
- Functions with side effects
- Functions that return different results for same input
- Functions that work with mutable objects
# Bad: Function with side effects
@cache # Don't do this!
def update_user_profile(user_id, data):
"""Update user profile in database."""
# This modifies state - shouldn't be cached
pass
# Bad: Non-deterministic function
@cache # Don't do this!
def get_random_number():
"""Generate random number."""
return random.random()
Memory Management
Monitor memory usage with large caches:
from functools import lru_cache
import sys
@lru_cache(maxsize=1000)
def memory_intensive_function(data):
"""Function that might use lots of memory."""
return [x * 2 for x in data]
def monitor_cache_memory(func):
"""Monitor cache memory usage."""
info = func.cache_info()
print(f"Cache size: {info.currsize}")
print(f"Cache hits: {info.hits}")
print(f"Cache misses: {info.misses}")
# Estimate memory usage (rough approximation)
if hasattr(func, '__wrapped__'):
cache_dict = func.__wrapped__.__self__
memory_estimate = sys.getsizeof(cache_dict)
print(f"Estimated cache memory: {memory_estimate} bytes")
# Clear cache when memory usage is too high
def manage_cache_memory(func, max_memory_mb=100):
"""Clear cache if estimated memory usage exceeds limit."""
info = func.cache_info()
if info.currsize > max_memory_mb * 1024 * 1024: # Convert MB to bytes
func.cache_clear()
print("Cache cleared due to memory limit")
Conclusion
functools.lru_cache
and cache
are powerful tools for performance optimization in Python. They provide:
- Dramatic performance improvements for expensive function calls
- Simple decorator syntax that's easy to apply
- Built-in cache management with LRU eviction
- Cache introspection with statistics and monitoring
Key takeaways:
- Use
@cache
for unlimited caching of pure functions - Use
@lru_cache(maxsize=N)
when memory usage is a concern - Monitor cache hit rates to ensure effectiveness
- Avoid caching functions with side effects or non-deterministic results
- Consider implementing time-based expiration for data that becomes stale
By mastering these caching techniques, you'll be able to significantly improve your Python application's performance while maintaining clean, readable code. Remember that caching is a trade-off between memory usage and computational speed - use it wisely for maximum benefit.
Add Comment
No comments yet. Be the first to comment!