Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
"""
Local/In-Process Cache
Pros: Fastest access, no network overhead
Cons: Not shared, limited by process memory
Use: Computation results, configuration
"""
return
"latency"
"1-10 μs"
"shared"
False
"persistence"
False
"examples"
"@lru_cache"
"dict"
"Redis in same host"
@staticmethod
def
distributed_cache
"""
Distributed Cache
Pros: Shared across instances, scalable
Cons: Network latency, complexity
Use: Session data, API responses
"""
return
"latency"
"1-10 ms"
"shared"
True
"persistence"
"optional"
"examples"
"Redis Cluster"
"Memcached"
@staticmethod
def
edge_cache
"""
Edge/CDN Cache
Pros: Global distribution, reduce origin load
Cons: Invalidation complexity
Use: Static assets, public content
"""
return
"latency"
"10-50 ms"
"shared"
True
"persistence"
True
"examples"
"Cloudflare"
"Fastly"
"CloudFront"
Caching Patterns
1. Cache-Aside (Lazy Loading)
Concept: Application checks cache first, loads from source on miss, then populates cache
Flow:
1. Check cache for data
2. If HIT → return cached data
3. If MISS → query source (DB/API)
4. Store result in cache
5. Return data
Implementation:
from functools import wraps
from typing importCallable, Anyimport time
classCacheAside:
"""Cache-Aside (Lazy Loading) pattern"""def__init__(self):
self.cache: Dict[str, tuple[Any, float]] = {}
self.ttl = 300# 5 minutesdefget(self, key: str) -> Optional[Any]:
"""Get from cache if not expired"""if key inself.cache:
value, timestamp = self.cache[key]
if time.time() - timestamp < self.ttl:
return value
else:
# Expireddelself.cache[key]
returnNonedefset(self, key: str, value: Any):
"""Set value in cache with timestamp"""self.cache[key] = (value, time.time())
defdecorator(self, key_func: Callable = None):
"""Decorator for cache-aside pattern"""defdecorator_wrapper(func):
@wraps(func)defwrapper(*args, **kwargs):
# Generate cache keyif key_func:
cache_key = key_func(*args, **kwargs)
else:
cache_key = f"{func.__name__}:{args}:{kwargs}"# Check cache
cached_value = self.get(cache_key)
if cached_value isnotNone:
return cached_value
# Cache miss - call function
result = func(*args, **kwargs)
# Store in cacheself.set(cache_key, result)
return result
return wrapper
return decorator_wrapper
# Usage
cache = CacheAside()
@cache.decorator(key_func=lambda user_id: f"user:{user_id}")defget_user(user_id: int):
"""Fetch user from database (expensive operation)"""print(f"Database query for user {user_id}")
# Simulate DB query
time.sleep(0.1)
return {"id": user_id, "name": f"User {user_id}"}
# First call - cache miss
user1 = get_user(1) # Prints "Database query..."# Second call - cache hit
user1_again = get_user(1) # No print, returns from cache
Pros:
Only caches requested data (efficient memory use)
Application controls caching logic
Resilient (cache failures don't block requests)
Cons:
Initial request latency (cache miss penalty)
Potential cache stampede on popular keys
Stale data during TTL window
2. Write-Through
Concept: Write to cache and source simultaneously
Flow:
1. Write data to cache
2. Write data to source (DB/API)
3. Return success only when both complete
Implementation:
classWriteThroughCache:
"""Write-Through caching pattern"""def__init__(self, database):
self.cache: Dict[str, Any] = {}
self.db = database
defget(self, key: str) -> Optional[Any]:
"""Read from cache, fall back to DB"""if key inself.cache:
returnself.cache[key]
# Cache miss - load from DB
value = self.db.get(key)
if value isnotNone:
self.cache[key] = value
return value
defset(self, key: str, value: Any):
"""Write to both cache and DB"""# Write to cache firstself.cache[key] = value
# Then write to databaseself.db.set(key, value)
# Both must succeed for consistency# UsageclassMockDatabase:
def__init__(self):
self.data = {}
defget(self, key):
returnself.data.get(key)
defset(self, key, value):
self.data[key] = value
db = MockDatabase()
cache = WriteThroughCache(db)
# Write
cache.set("user:1", {"name": "Alice"})
# Cache and DB both updated# Read
user = cache.get("user:1") # From cache
Pros:
Data consistency (cache always matches source)
No cache miss penalty on reads
Simplifies cache warming
Cons:
Write latency (two operations)
Wasted cache space (all writes cached, even if never read)
Write failures affect both layers
3. Write-Behind (Write-Back)
Concept: Write to cache immediately, asynchronously write to source
Flow:
1. Write data to cache
2. Return success immediately
3. Asynchronously batch writes to source
Implementation:
import asyncio
from collections import deque
from dataclasses import dataclass
from typing import Deque
@dataclassclassWriteOperation:
"""Pending write operation"""
key: str
value: Any
timestamp: floatclassWriteBehindCache:
"""Write-Behind (Write-Back) caching pattern"""def__init__(self, database, batch_size=10, flush_interval=5.0):
self.cache: Dict[str, Any] = {}
self.db = database
self.write_queue: Deque[WriteOperation] = deque()
self.batch_size = batch_size
self.flush_interval = flush_interval
self.running = Falseasyncdefstart(self):
"""Start background flushing"""self.running = Truewhileself.running:
await asyncio.sleep(self.flush_interval)
awaitself.flush()
asyncdefflush(self):
"""Flush pending writes to database"""ifnotself.write_queue:
return# Batch writes
batch = []
whileself.write_queue andlen(batch) < self.batch_size:
batch.append(self.write_queue.popleft())
# Write batch to databasefor op in batch:
self.db.set(op.key, op.value)
print(f"Flushed {len(batch)} writes to database")
defget(self, key: str) -> Optional[Any]:
"""Read from cache"""returnself.cache.get(key)
defset(self, key: str, value: Any):
"""Write to cache, queue for DB write"""# Immediate write to cacheself.cache[key] = value
# Queue for async DB writeself.write_queue.append(
WriteOperation(key, value, time.time())
)
Pros:
Fast writes (no wait for DB)
Batching reduces DB load
Better write throughput
Cons:
Risk of data loss (if cache crashes before flush)
Complexity (background jobs, retry logic)
Eventual consistency
4. Read-Through
Concept: Cache automatically loads data from source on miss
Implementation:
classReadThroughCache:
"""Read-Through caching pattern"""def__init__(self, loader_func: Callable[[str], Any]):
self.cache: Dict[str, Any] = {}
self.loader = loader_func # Function to load from sourcedefget(self, key: str) -> Any:
"""
Get value, automatically loading on miss
Cache handles loading - application doesn't know about source
"""if key inself.cache:
returnself.cache[key]
# Cache miss - load from source
value = self.loader(key)
# Store in cacheself.cache[key] = value
return value
# Usagedefload_user_from_db(user_id: str):
"""Loader function"""print(f"Loading user {user_id} from database")
return {"id": user_id, "name": f"User {user_id}"}
cache = ReadThroughCache(loader_func=load_user_from_db)
# Application doesn't handle cache misses
user = cache.get("user:1") # Automatically loads if not cached
Eviction Policies
LRU (Least Recently Used)
Concept: Evict least recently accessed item when cache is full
Implementation:
from collections import OrderedDict
classLRUCache:
"""LRU Cache with O(1) get and set"""def__init__(self, capacity: int):
self.capacity = capacity
self.cache = OrderedDict()
defget(self, key: str) -> Optional[Any]:
"""Get value and mark as recently used"""if key notinself.cache:
returnNone# Move to end (most recent)self.cache.move_to_end(key)
returnself.cache[key]
defset(self, key: str, value: Any):
"""Set value, evict LRU if at capacity"""if key inself.cache:
# Update existingself.cache.move_to_end(key)
else:
# New keyiflen(self.cache) >= self.capacity:
# Evict least recently used (first item)self.cache.popitem(last=False)
self.cache[key] = value
def__len__(self):
returnlen(self.cache)
# Usage
lru = LRUCache(capacity=3)
lru.set("a", 1)
lru.set("b", 2)
lru.set("c", 3)
lru.get("a") # Access 'a', now most recent
lru.set("d", 4) # Evicts 'b' (least recently used)print("b"in lru.cache) # Falseprint("a"in lru.cache) # True
Use when: Access patterns favor recent items (temporal locality)
LFU (Least Frequently Used)
Concept: Evict least frequently accessed item
Implementation:
from collections import defaultdict
import heapq
classLFUCache:
"""LFU Cache with frequency tracking"""def__init__(self, capacity: int):
self.capacity = capacity
self.cache: Dict[str, Any] = {}
self.frequency: Dict[str, int] = defaultdict(int)
self.access_time: Dict[str, int] = {}
self.time = 0defget(self, key: str) -> Optional[Any]:
"""Get value and increment frequency"""if key notinself.cache:
returnNoneself.frequency[key] += 1self.time += 1self.access_time[key] = self.time
returnself.cache[key]
defset(self, key: str, value: Any):
"""Set value, evict LFU if at capacity"""ifself.capacity == 0:
returnif key inself.cache:
self.cache[key] = value
self.frequency[key] += 1self.time += 1self.access_time[key] = self.time
returniflen(self.cache) >= self.capacity:
# Find least frequently used# Break ties by least recently used
lfu_key = min(
self.cache.keys(),
key=lambda k: (self.frequency[k], self.access_time[k])
)
delself.cache[lfu_key]
delself.frequency[lfu_key]
delself.access_time[lfu_key]
self.cache[key] = value
self.frequency[key] = 1self.time += 1self.access_time[key] = self.time
Use when: Some items accessed much more frequently than others
TTL (Time To Live)
Concept: Evict items after fixed time period
import time
classTTLCache:
"""TTL-based cache with automatic expiration"""def__init__(self, default_ttl: float = 300):
self.cache: Dict[str, tuple[Any, float]] = {}
self.default_ttl = default_ttl
defget(self, key: str) -> Optional[Any]:
"""Get value if not expired"""if key notinself.cache:
returnNone
value, expiry = self.cache[key]
if time.time() > expiry:
# Expireddelself.cache[key]
returnNonereturn value
defset(self, key: str, value: Any, ttl: Optional[float] = None):
"""Set value with TTL"""if ttl isNone:
ttl = self.default_ttl
expiry = time.time() + ttl
self.cache[key] = (value, expiry)
defcleanup(self):
"""Remove all expired entries"""
now = time.time()
expired = [k for k, (_, exp) inself.cache.items() if now > exp]
for k in expired:
delself.cache[k]
Use when: Data has natural expiration (sessions, temporary tokens)
Cache Key Design
Best Practices
classCacheKeyDesign:
"""Cache key naming best practices""" @staticmethoddefhierarchical_key(namespace: str, entity: str, id: str) -> str:
"""
Hierarchical naming for organization
Pattern: namespace:entity:id
Example: app:user:123, api:product:456
"""returnf"{namespace}:{entity}:{id}" @staticmethoddefcomposite_key(*parts) -> str:
"""
Composite key from multiple values
Example: user_posts(user_id, page) → "posts:user:123:page:1"
"""return":".join(str(p) for p in parts)
@staticmethoddefhash_key(data: str) -> str:
"""
Hash long or complex keys
Use for: Query strings, JSON, URLs
"""import hashlib
return hashlib.sha256(data.encode()).hexdigest()[:16]
@staticmethoddefversion_key(key: str, version: int) -> str:
"""
Versioned keys for invalidation
Increment version to invalidate all old keys
"""returnf"{key}:v{version}"# Examples
keys = CacheKeyDesign()
# User data
user_key = keys.hierarchical_key("app", "user", "123")
# "app:user:123"# Paginated results
posts_key = keys.composite_key("posts", "user", 123, "page", 1)
# "posts:user:123:page:1"# Complex query
query = "SELECT * FROM users WHERE age > 25 AND city = 'NYC'"
query_key = f"query:{keys.hash_key(query)}"# Versioned cache
config_key = keys.version_key("app:config", version=2)
# "app:config:v2"
When to Cache vs Not Cache
Cache These
classGoodCacheCandidates:
"""Data that benefits from caching"""
EXAMPLES = {
"Expensive computations": {
"example": "ML model inference, complex calculations",
"ttl": "hours to days",
"pattern": "Cache-Aside"
},
"Frequently accessed data": {
"example": "User profiles, product catalogs",
"ttl": "minutes to hours",
"pattern": "Read-Through"
},
"Slow external API calls": {
"example": "Third-party APIs, microservices",
"ttl": "minutes",
"pattern": "Cache-Aside"
},
"Static or rarely changing": {
"example": "Configuration, reference data",
"ttl": "hours to days",
"pattern": "Write-Through"
},
"High read-to-write ratio": {
"example": "News articles, blog posts",
"ttl": "minutes to hours",
"pattern": "Cache-Aside"
}
}
Don't Cache These
classPoorCacheCandidates:
"""Data that should NOT be cached"""
EXAMPLES = [
"Highly personalized data (unless user-keyed)",
"Rapidly changing data (stock prices, live scores)",
"Large objects (>1MB, unless CDN)",
"Data accessed once (no reuse benefit)",
"Security-sensitive data (PII, passwords)",
"Already fast queries (<10ms)",
]
Common Anti-Patterns
❌ No Cache Expiration
# WRONG: Cache lives forever
cache = {}
cache[key] = value # Never expires# CORRECT: Set TTL
cache.set(key, value, ttl=300)
❌ Caching Failures
# WRONG: Cache error responses
result = api_call()
cache.set(key, result) # What if result is error?# CORRECT: Only cache successful responses
result = api_call()
if result.success:
cache.set(key, result)
❌ Cache Stampede
# WRONG: All requests miss simultaneously# (e.g., cache expires at exact time)# CORRECT: Probabilistic early expirationimport random
defget_with_early_expiration(key, ttl):
value, expiry = cache.get_with_expiry(key)
# Probabilistically refresh before expiry
time_left = expiry - time.time()
if time_left < ttl * random.random():
# Refresh cache
value = fetch_fresh_data(key)
cache.set(key, value, ttl)
return value
Quick Reference
Cache Pattern Selection
Pattern
Read Speed
Write Speed
Consistency
Use Case
Cache-Aside
Medium (miss penalty)
Fast
Eventual
General purpose, read-heavy
Write-Through
Fast
Slow
Strong
Consistent reads required
Write-Behind
Fast
Very Fast
Eventual
High write throughput
Read-Through
Fast
N/A
Eventual
Simplified read logic
Eviction Policy Selection
Policy
Best For
Worst For
LRU
Temporal locality
Scanning workloads
LFU
Skewed access patterns
Changing patterns
TTL
Time-sensitive data
Static data
FIFO
Fair eviction
Performance optimization
Related Skills
Next Steps:
http-caching.md → Browser and HTTP cache layer
cdn-edge-caching.md → CDN and edge caching
redis-caching-patterns.md → Distributed caching with Redis
