Caching & Redis
Cache-aside, TTL, invalidation, hot keys, and in-memory lookup patterns at scale.
Cache Penetration Null Object Mock in Python
Implement a cache that stores a null marker on misses to prevent repeated database hits, reducing cache penetration.
import time
from collections import defaultdict
from typing import Any, Optional
class Cache:
def __init__(self):
self.store: dict[str, Any] = {}
self.ttl: dict[str, float] = {}
self.null_marker = object()
def get(self, key: str, ttl: int = 60, fallback:
Any = None) -> An…How to Add TTL Jitter to Cache Expiration in Python
A Python decorator that adds random jitter to cache TTLs, staggering expiration times to prevent cache avalanche.
import random
import time
from functools import wraps
def add_jitter(ttl: float, jitter_range: float = 0.1) -> float:
"""Add random jitter (as % of TTL) to stagger cache expiration and prevent avalanche."""
jitter = random.uniform(-jitter_range, jitter_range)
return ttl * (1 + jitter)
def cache_with_jitt…How to Implement Probabilistic Early Expiration in Python
A Python mock of probabilistic early expiration for caches, using a heap-based expiry queue and random eviction to approximate cache stampede protection.
import heapq
import random
import time
class ProbabilisticEarlyExpirationMock:
def __init__(self, capacity=1024, expiration_probability=0.1):
self.capacity = capacity
self.expiration_probability = expiration_probability
self._items = {}
self._expiry_heap = []
self._next_id…How to Implement a Negative Cache with TTL in Python
This code provides a TTL mock cache that stores negative results (cache misses) for a short time to reduce repeated lookups of missing keys.
from time import time, sleep
class TTLMockCache:
def __init__(self, ttl_seconds=5):
self.ttl = ttl_seconds
self.store = {}
self.negative_cache = {}
def get(self, key):
now = time()
if key in self.store:
value, expires_at = self.store[key]
if exp…How to Implement a Redis-Like Cache Dictionary in Python
Build a RedisMockDict class that mimics basic Redis key-value operations with TTL support, expiry cleanup, and standard dict-like methods.
from collections import OrderedDict
import time
class RedisMockDict:
def __init__(self, ttl=None):
self._data = OrderedDict()
self._ttl = ttl # default TTL in seconds, None = no expiry
self._expiry = {}
def set(self, key, value, ttl=None):
"""Set a key-value pair with optiona…How to Implement a Write-Through Cache in Python with a Mock Database
A thread-safe write-through cache that updates both cache and mock database atomically, computing values only after a successful write to the database.
import threading
import time
import random
class WriteThroughCache:
def __init__(self):
self.cache = {}
self.db = {}
self.lock = threading.Lock()
def write(self, key, value):
with self.lock:
# Simulate slow database write
time.sleep(random.uniform(0.01…How to Mock Cache Tag Invalidation in Python
Use unittest.mock.patch with wraps to verify tagged cache entries are invalidated correctly.
import unittest
from unittest.mock import patch
def get_cached_data(cache, key):
"""Return data from cache if present and valid, else None."""
if cache.get(key, {}).get("valid", False):
return cache[key]["data"]
return None
def invalidate_tag_mock(cache, tag):
"""Invalidate all cache entries …How to Mock Redis EXPIRE, TTL, and PERSIST in Python
A lightweight in-memory MockRedis class that simulates Redis key expiration, TTL, and persist behavior for tests and local development.
import time
class MockRedis:
def __init__(self):
self._store = {}
self._expiry = {}
def set(self, key, value):
self._store[key] = value
self._expiry.pop(key, None)
return True
def expire(self, key, ttl_seconds):
if key not in self._store:
retur…How to Mock Redis Pipeline Batch Commands in Python
Create a lightweight MockRedis class that simulates Redis pipeline batching with SET, GET, and DELETE operations for testing without a live server.
import redis
import time
class MockRedis:
def __init__(self):
self.data = {}
def pipeline(self):
return MockPipeline(self)
def execute(self, commands):
results = []
for cmd in commands:
op, args = cmd[0], cmd[1:]
if op == "SET":
se…How to Mock Redis Pub/Sub in Python
Test Redis pub/sub logic without a live server using an in-memory fake that queues published messages per channel.
import redis
import time
import threading
class MockRedisPubSub:
def __init__(self):
self.channels = {}
def publish(self, channel, message):
if channel not in self.channels:
return 0
for subscriber in self.channels[channel]:
subscriber.put(message)
ret…How to Mock Redis Streams Consumer Groups in Python
Simulate Redis Streams producer and consumer group behavior in Python using a standalone mock class for testing and development.
import time
import json
from collections import defaultdict
class RedisStreamMock:
def __init__(self):
self.streams = defaultdict(list)
self.consumer_groups = defaultdict(dict)
self.pending_entries = defaultdict(list)
def xadd(self, stream, fields):
entry_id = f"{time.time_ns(…How to Mock a Redis Session Store Cookie SID in Python
Mock a Redis-backed session store with a cookie-based session ID (SID) in Python, including the create, read, and delete operations.
import redis
import uuid
import time
class RedisSessionStore:
def __init__(self, host="localhost", port=6379, db=0, prefix="session:"):
self.client = redis.Redis(host=host, port=port, db=db)
self.prefix = prefix
def create_session(self, timeout_seconds=3600):
session_id = uuid.uuid4(…How to Mock a Redis Transaction with MULTI/EXEC in Python
A minimal in-memory mock of Redis MULTI/EXEC transactions that queues commands and applies them atomically on EXEC.
class RedisTransactionMock:
def __init__(self):
self.data = {}
self.queue = []
self.in_transaction = False
def multi(self):
self.in_transaction = True
self.queue = []
return "OK"
def set(self, key, value):
if self.in_transaction:
self.qu…How to Mock zlib Compression for Cache Values in Python
Compress cache values with zlib and mock the compress function in unit tests to simulate cache behavior.
import zlib
from unittest.mock import patch
def compress_value(data: bytes) -> bytes:
"""Compress data using zlib and return the compressed bytes."""
return zlib.compress(data)
def decompress_value(compressed: bytes) -> bytes:
"""Decompress zlib data and return the original bytes."""
return zlib.deco…How to Serialize Cache Values with JSON and Pickle in Python
Serialize cache values using JSON for simple types or pickle for arbitrary objects, with robust error handling for unsupported types like mocks.
import json
import pickle
from unittest.mock import Mock
def serialize(value, method="json"):
"""Serialize a cache value using JSON or pickle with type checking."""
if method == "json":
try:
return json.dumps(value).encode("utf-8")
except TypeError as e:
raise ValueErro…How to implement Redlock distributed lock in Python
Simulate Redis Redlock multi-instance locking to show how a distributed lock is acquired only when a majority of instances agree.
import time
import random
import threading
from dataclasses import dataclass
@dataclass
class MockRedisLock:
"""Simple mock of a Redis lock instance."""
name: str
key: str
ttl: int
acquired: bool = False
expires_at: float = 0.0
def acquire(self, sleep_fn=time.sleep):
"""Try to ac…How to mock Redis geospatial commands (GEOADD) in Python
Implement a lightweight Python mock of Redis geospatial commands (GEOADD, GEODIST, GEOSEARCH) using the Haversine formula for testing without a Redis server.
import math
import heapq
class MockRedisGeo:
def __init__(self):
self.members = {}
def geoadd(self, key, longitude, latitude, member):
if key not in self.members:
self.members[key] = {}
self.members[key][member] = (longitude, latitude)
def geodist(self, key, member1,…Implement a Multi-Level Cache with L1 Memory and L2 Redis in Python
This code implements a simple multi-level cache with an in-process L1 cache (via functools.lru_cache) and a mock Redis L2 cache with TTL, falling back to a slow computation on misses.
import time
from functools import lru_cache
class MockRedis:
def __init__(self):
self.store = {}
def get(self, key):
return self.store.get(key, None)
def set(self, key, value, ttl=5):
self.store[key] = (value, time.time() + ttl)
def get_ttl(self, key):
value, expiry…Implement a TTL cache with a mock clock in Python
This code creates a simple TTL cache that stores values with an expiration timestamp and allows injecting a mock time function to test expiry behavior deterministically.
import time
from functools import wraps
class TTLCache:
def __init__(self, ttl_seconds):
self.ttl = ttl_seconds
self.cache = {}
self._now = time.time
def set_mock_time(self, mock_time_fn):
"""Inject a mock time function for testing TTL expiry."""
self._now = mock_time_…Mock Redis Distributed Lock in Python with SET NX EX
A minimal in-memory mock of Redis SET NX EX distributed lock semantics for testing concurrent code without a real Redis server.
import time
import threading
import uuid
from typing import Optional
class RedisLockMock:
"""A minimal mock of Redis SET NX EX distributed lock semantics."""
def __init__(self):
self._store = {} # key -> (value, expiry_epoch)
def acquire(self, key: str, token: str, ttl_seconds: int) -> bool:
…Python Redis WATCH optimistic lock mock
A MockRedis class that simulates WATCH/MULTI/EXEC transactions with optimistic locking to detect concurrent modifications before committing.
import time
import threading
class MockRedis:
def __init__(self):
self.data = {}
self.watched = {}
self.lock = threading.Lock()
def get(self, key):
return self.data.get(key)
def set(self, key, value):
self.data[key] = value
def watch(self, *keys):
wi…Redis Leaky Bucket Rate Limiting Mock in Python
Simulates a Redis-backed leaky bucket rate limiter using a local class with continuous leaking and token capacity checks.
import time
from collections import deque
class LeakyBucket:
def __init__(self, capacity, leak_rate):
self.capacity = capacity
self.leak_rate = leak_rate
self.water = 0.0
self.timestamp = time.time()
self.history = deque()
def allow(self):
current = time.time(…Redis-inspired sliding window rate limiter in Python
A pure-Python sliding window rate limiter using a deque of timestamps, mock-ready for Redis-backed production limits.
import time
from collections import deque
class SlidingWindowRateLimiter:
def __init__(self, max_requests: int, window_seconds: int) -> None:
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests: dict[str, deque] = {}
def is_allowed(self, client_id: str…Refresh Proactive TTL Renewal in Python
This snippet implements a proactive TTL renewal pattern that refreshes a cache expiration before it lapses, using a mock counter to track renewals.
import time
from datetime import datetime, timezone
class TTLRenewer:
def __init__(self, ttl_seconds=10, renew_at=0.5):
self.ttl = ttl_seconds
self.last_renewed = time.time()
self.renew_threshold = ttl_seconds * renew_at
self.renewals = 0
def check_and_renew(self):
if …Browse by section
Each section groups closely related Python snippets.
Caching & Redis — Python code examples
What you will find here
This page collects caching & redis snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.
Samples vs tutorials and challenges
Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.