Background: When the Database Becomes a Bottleneck
I once ran into a pretty common problem: a Python script processing 100K records, where each record required a database query to fetch configuration data. The result? The script took over 20 minutes to run, and the database CPU spiked to 90%. The problem wasn’t in the processing logic — it was that the same query was being called thousands of times, returning identical results every single time.
Redis and redis-py are what saved that script. Instead of hitting the database every time data needed to be read, Redis holds results in memory — read/write speeds measured in microseconds rather than milliseconds. That 20-minute script dropped to 90 seconds after adding a cache layer. Not by rewriting the algorithm, but simply by avoiding doing the same work over and over.
This article focuses on the 3 techniques I use most frequently with redis-py: Caching (eliminating redundant queries), Pipelines (batching commands to reduce network round-trips), and Distributed Locks (ensuring consistency when multiple processes run in parallel).
Installing redis-py
First, you need a running Redis server. The quickest way is to use Docker:
docker run -d --name redis-local -p 6379:6379 redis:7-alpineThen install the Python library:
pip install redis
# Install hiredis for faster response parsing
pip install redis[hiredis]hiredis is a C extension that speeds up response parsing — there’s no reason to skip it. Check the connection:
import redis
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
print(r.ping()) # True if connection is successfulNote the decode_responses=True — without this parameter, all returned values are bytes. I used to forget this when first starting out and would spend 10 minutes debugging why string comparisons weren’t working.
Detailed Configuration
1. Caching with Cache-Aside Pattern
The idea is fairly intuitive: read from Redis first, and only fall back to the DB on a cache miss — then save the result to Redis with a TTL so the next read doesn’t need to hit the DB again.
import redis
import json
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
def get_user_config(user_id: int) -> dict:
cache_key = f"user:config:{user_id}"
# Read from cache first
cached = r.get(cache_key)
if cached:
return json.loads(cached)
# Cache miss — read from DB
config = fetch_from_database(user_id) # Your actual DB query function
# Save to cache with 5-minute TTL
r.setex(cache_key, 300, json.dumps(config))
return configWith 100K records but only 500 unique user_ids, DB queries drop from 100K to ~500 — 200 times fewer. Database CPU goes from 90% down to around 15%, and the script from 20 minutes to 90 seconds. A few dozen extra lines of code, no refactoring required.
If you don’t want to write boilerplate for every function, you can wrap it with a decorator:
from functools import wraps
def redis_cache(ttl=300):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
key = f"cache:{func.__name__}:{args}:{sorted(kwargs.items())}"
result = r.get(key)
if result:
return json.loads(result)
result = func(*args, **kwargs)
r.setex(key, ttl, json.dumps(result))
return result
return wrapper
return decorator
@redis_cache(ttl=600)
def get_product_details(product_id: int) -> dict:
# Query database...
pass2. Pipelines — Reducing Network Round-trips
Every Redis command sent is a network round-trip. If you need to run 1,000 individual commands over a network with 1ms latency, that’s 1 second spent purely on network overhead. Pipelines batch multiple commands into a single request, sending them all at once and receiving all results at once.
import redis
import time
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# Without pipeline (slow)
def set_scores_naive(scores: dict):
start = time.perf_counter()
for user_id, score in scores.items():
r.set(f"score:{user_id}", score)
print(f"Naive: {time.perf_counter() - start:.3f}s")
# Using Pipeline (8-12x faster)
def set_scores_pipeline(scores: dict):
start = time.perf_counter()
with r.pipeline(transaction=False) as pipe: # transaction=False for higher throughput
for user_id, score in scores.items():
pipe.set(f"score:{user_id}", score)
pipe.execute() # Send all at once
print(f"Pipeline: {time.perf_counter() - start:.3f}s")
# Test with 10,000 records
test_scores = {f"user_{i}": i * 10 for i in range(10000)}
set_scores_naive(test_scores) # ~2.5s on localhost
set_scores_pipeline(test_scores) # ~0.3s on localhostPipelines are also useful when reading multiple keys at once:
def get_multiple_users(user_ids: list) -> dict:
with r.pipeline(transaction=False) as pipe:
for uid in user_ids:
pipe.get(f"user:config:{uid}")
results = pipe.execute()
return {
uid: json.loads(res) if res else None
for uid, res in zip(user_ids, results)
}transaction=False disables MULTI/EXEC wrapping to increase throughput. Only enable transaction=True when you genuinely need atomic execution — for example, when reading and then writing based on the value you just read.
3. Distributed Locks — Preventing Race Conditions
A common real-world scenario: 3 worker processes all check “has the email been sent?” and all decide “no, I’ll send it.” The result is the user receiving 3 identical emails. A distributed lock prevents this by ensuring only one process executes a critical section at any given time.
import redis
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
def send_notification_once(user_id: int, message: str) -> bool:
lock_key = f"lock:notification:{user_id}"
# Try to acquire lock with 10-second timeout
lock = r.lock(lock_key, timeout=10, blocking_timeout=5)
acquired = lock.acquire(blocking=True)
if not acquired:
print(f"Could not acquire lock for user {user_id}, skipping")
return False
try:
# Double-check after acquiring lock
sent_key = f"notification:sent:{user_id}"
if r.exists(sent_key):
return False
# Send notification
send_email(user_id, message)
# Mark as sent, TTL 1 hour
r.setex(sent_key, 3600, "1")
return True
finally:
lock.release() # ALWAYS release in finallylock.release() must be in the finally block — no exceptions. If an exception is raised without releasing the lock, all other processes will be blocked until the timeout expires on its own. Two parameters to keep in mind: timeout is how long the lock auto-expires if the process crashes (a safety net), and blocking_timeout is the maximum time a process will wait to acquire the lock.
Another common use case is preventing cron jobs from running in parallel:
def run_scheduled_job(job_name: str):
lock_key = f"job:running:{job_name}"
lock = r.lock(lock_key, timeout=300) # Job runs for at most 5 minutes
if not lock.acquire(blocking=False): # Non-blocking: skip if already running
print(f"Job {job_name} is running in another worker, skipping")
return
try:
execute_job(job_name)
finally:
lock.release()Testing & Monitoring
Cache hit rate
Redis tracks hits and misses directly in memory — no external tools needed. This command gives you the numbers immediately:
redis-cli info stats | grep -E "keyspace_hits|keyspace_misses"
# Sample output:
# keyspace_hits:45231
# keyspace_misses:1823
# Hit rate = 45231 / (45231 + 1823) ≈ 96%def get_cache_stats() -> dict:
info = r.info('stats')
hits = info['keyspace_hits']
misses = info['keyspace_misses']
total = hits + misses
hit_rate = (hits / total * 100) if total > 0 else 0
return {
'hits': hits,
'misses': misses,
'hit_rate': f"{hit_rate:.1f}%",
}
def get_memory_usage() -> dict:
info = r.info('memory')
return {
'used': info['used_memory_human'],
'fragmentation_ratio': info['mem_fragmentation_ratio'],
}Slow Log and Memory Configuration
# Enable slow log — log commands slower than 1ms
redis-cli config set slowlog-log-slower-than 1000 # microseconds
redis-cli slowlog get 10
# Limit memory, auto-evict old keys using LRU
redis-cli config set maxmemory 512mb
redis-cli config set maxmemory-policy allkeys-lruThe allkeys-lru policy suits most caching use cases: Redis automatically evicts the least recently used keys when memory is full, instead of returning an OOM error. Setting maxmemory is a step you can never skip before going to production.
Real-world Pipeline Benchmark
import time
def benchmark_pipeline(n=5000):
test_data = {f"key:{i}": f"value:{i}" for i in range(n)}
start = time.perf_counter()
for k, v in test_data.items():
r.set(k, v)
t_normal = time.perf_counter() - start
start = time.perf_counter()
with r.pipeline(transaction=False) as pipe:
for k, v in test_data.items():
pipe.set(k, v)
pipe.execute()
t_pipeline = time.perf_counter() - start
print(f"Normal: {t_normal:.3f}s")
print(f"Pipeline: {t_pipeline:.3f}s")
print(f"Speedup: {t_normal / t_pipeline:.1f}x")
benchmark_pipeline()On localhost (Redis 7, Python 3.11), 5,000 operations run 8–12x faster. But when connecting over a real network with 10–50ms latency, the difference is even more dramatic — 1,000 commands × 10ms = 10 seconds of overhead. With a pipeline, all 1,000 of those commands are still just a single round-trip.
These three techniques are independent — use whichever fits the problem at hand. Caching when you notice abnormally high database CPU. Pipelines when you need bulk reads or writes. Distributed locks when multiple workers are processing the same data stream. Apply them in the right place, and redis-py will rarely let you down.
