Understanding the Redis Cache Layer for PostgreSQL
PostgreSQL is a powerful relational database, but even the most optimized queries can become bottlenecks under heavy load. A Redis cache layer sits between your application and PostgreSQL, storing frequently accessed data in memory to dramatically reduce database load and improve response times. Redis, as an in-memory data store, offers sub-millisecond latency compared to the often multi-millisecond or even second-long queries hitting disk-backed databases.
This tutorial walks you through building a production-ready cache layer using the Cache-Aside pattern (also called Lazy Loading), where the application first checks Redis, and only queries PostgreSQL on a cache miss, then backfills the cache for future requests.
Why a Cache Layer Matters
Without caching, every API request that needs user profiles, product details, or session data hits PostgreSQL directly. This leads to several problems:
- Connection exhaustion — PostgreSQL has a finite pool of connections; too many concurrent queries can saturate it
- Repeated expensive queries — JOINs, aggregations, and full-text searches run over and over for the same data
- Latency spikes — Disk I/O and query planning add up, pushing p95 latencies into unacceptable ranges
- Database contention — Read-heavy workloads can still create lock contention on frequently accessed rows
Introducing Redis changes the equation. A well-configured cache layer can handle 80-95% of reads from memory, letting PostgreSQL focus on writes and complex transactional work. The result is lower infrastructure costs, faster user experiences, and a system that scales horizontally with ease.
Architecture Overview: The Cache-Aside Pattern
The Cache-Aside pattern follows a simple flow:
- Read: Application checks Redis → On hit, return cached data → On miss, query PostgreSQL → Store result in Redis with a TTL → Return data
- Write: Application writes to PostgreSQL directly → Invalidates (deletes) the corresponding Redis key → On next read, the fresh data is loaded from PostgreSQL and cached
This pattern ensures the database is always the source of truth, while Redis serves as a fast, disposable read replica. The TTL (Time-To-Live) on keys prevents stale data from persisting indefinitely.
Step 1: Setting Up the Environment
You'll need a running PostgreSQL instance and a Redis instance. For local development, Docker is the easiest path:
# Start PostgreSQL
docker run -d --name pg-cache-demo \
-e POSTGRES_PASSWORD=secret \
-e POSTGRES_DB=appdb \
-p 5432:5432 \
postgres:16
# Start Redis
docker run -d --name redis-cache-demo \
-p 6379:6379 \
redis:7 --requirepass myredispassword
# Create a sample table in PostgreSQL
docker exec -i pg-cache-demo psql -U postgres -d appdb <<EOF
CREATE TABLE users (
id SERIAL PRIMARY KEY,
username TEXT NOT NULL UNIQUE,
email TEXT NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW()
);
-- Insert sample data
INSERT INTO users (username, email) VALUES
('alice', 'alice@example.com'),
('bob', 'bob@example.com'),
('charlie', 'charlie@example.com');
EOF
Install the required Python packages:
pip install redis psycopg2-binaryStep 2: Building the Core Cache Module
Create a dedicated cache module that encapsulates all Redis operations. This keeps your caching logic centralized and testable. The module will handle serialization, connection management, and the fundamental get/set/delete operations.
# cache_layer.py
import json
from datetime import timedelta
import redis
from typing import Optional, Any, Callable
class CacheLayer:
"""Redis cache layer with serialization and error handling."""
def __init__(self, redis_client: redis.Redis, default_ttl: int = 300):
self.redis = redis_client
self.default_ttl = default_ttl # 5 minutes default
def get(self, key: str) -> Optional[Any]:
"""Retrieve a value from cache. Returns None on miss or error."""
try:
raw = self.redis.get(key)
if raw is None:
return None
return json.loads(raw)
except (redis.RedisError, json.JSONDecodeError):
# On any Redis error, treat as a cache miss — don't crash
return None
def set(self, key: str, value: Any, ttl: Optional[int] = None) -> bool:
"""Store a value in cache with an optional TTL in seconds."""
try:
serialized = json.dumps(value, default=str)
expire = ttl if ttl is not None else self.default_ttl
return self.redis.setex(key, expire, serialized)
except redis.RedisError:
# Log the error in production, but don't block the application
return False
def delete(self, key: str) -> bool:
"""Remove a key from cache (used on write invalidation)."""
try:
return bool(self.redis.delete(key))
except redis.RedisError:
return False
def delete_pattern(self, pattern: str) -> int:
"""Delete all keys matching a glob pattern. Useful for cache groups."""
try:
keys = self.redis.keys(pattern)
if keys:
return self.redis.delete(*keys)
return 0
except redis.RedisError:
return 0
This wrapper is critical. It catches Redis errors gracefully — if Redis is temporarily unavailable, the application falls back to PostgreSQL without throwing exceptions. It also handles JSON serialization consistently so every cached item follows the same encoding.
Step 3: Implementing the Cache-Aside Read
The core pattern: try Redis first, fall back to PostgreSQL, then populate Redis for subsequent requests. Below is a complete example for fetching a user by ID:
# user_repository.py
import psycopg2
from psycopg2 import pool
from cache_layer import CacheLayer
# Connection pool for PostgreSQL
db_pool = pool.SimpleConnectionPool(
minconn=2,
maxconn=10,
host="localhost",
port=5432,
dbname="appdb",
user="postgres",
password="secret"
)
class UserRepository:
def __init__(self, cache: CacheLayer):
self.cache = cache
def _get_db_connection(self):
return db_pool.getconn()
def _release_db_connection(self, conn):
db_pool.putconn(conn)
def get_user_by_id(self, user_id: int) -> dict | None:
# Step 1: Build the cache key
cache_key = f"user:{user_id}"
# Step 2: Try Redis first
cached = self.cache.get(cache_key)
if cached is not None:
print(f" ✓ Cache HIT for user:{user_id}")
return cached
# Step 3: Cache miss — query PostgreSQL
print(f" ✗ Cache MISS for user:{user_id} — querying PostgreSQL")
conn = self._get_db_connection()
try:
with conn.cursor() as cur:
cur.execute(
"SELECT id, username, email, created_at FROM users WHERE id = %s",
(user_id,)
)
row = cur.fetchone()
if row is None:
return None
user = {
"id": row[0],
"username": row[1],
"email": row[2],
"created_at": row[3].isoformat()
}
# Step 4: Backfill cache (fire-and-forget — don't block on failure)
self.cache.set(cache_key, user, ttl=600) # 10 minutes
return user
finally:
self._release_db_connection(conn)
def get_user_by_username(self, username: str) -> dict | None:
"""Cache by username as well, with a different key namespace."""
cache_key = f"user:username:{username.lower()}"
cached = self.cache.get(cache_key)
if cached is not None:
return cached
conn = self._get_db_connection()
try:
with conn.cursor() as cur:
cur.execute(
"SELECT id, username, email, created_at FROM users WHERE username = %s",
(username,)
)
row = cur.fetchone()
if row is None:
return None
user = {
"id": row[0],
"username": row[1],
"email": row[2],
"created_at": row[3].isoformat()
}
# Cache under both ID and username keys for future lookups
self.cache.set(cache_key, user, ttl=600)
self.cache.set(f"user:{user['id']}", user, ttl=600)
return user
finally:
self._release_db_connection(conn)
Notice the dual caching in get_user_by_username — when you look up by username, you also cache under the user ID key. This way, future requests by ID won't miss. Multi-key backfill is a powerful technique that reduces cold-start cache misses.
Step 4: Handling Writes with Cache Invalidation
When data changes, the cached copy becomes stale. The safest approach is invalidation on write: delete the affected cache keys after a successful database write. The next read will repopulate from the fresh PostgreSQL data.
def update_user_email(self, user_id: int, new_email: str) -> bool:
"""Update user email and invalidate related cache keys."""
conn = self._get_db_connection()
try:
with conn.cursor() as cur:
# First, fetch the username so we can invalidate the username-keyed cache
cur.execute("SELECT username FROM users WHERE id = %s", (user_id,))
row = cur.fetchone()
if row is None:
return False
username = row[0]
# Perform the update in PostgreSQL (source of truth)
cur.execute(
"UPDATE users SET email = %s WHERE id = %s",
(new_email, user_id)
)
conn.commit()
# Invalidate both cache keys — the data has changed
self.cache.delete(f"user:{user_id}")
self.cache.delete(f"user:username:{username.lower()}")
print(f" ✓ Updated user:{user_id} — invalidated cache keys")
return True
except Exception:
conn.rollback()
raise
finally:
self._release_db_connection(conn)
def create_user(self, username: str, email: str) -> dict | None:
"""Create a new user. No cache invalidation needed, but we can warm the cache."""
conn = self._get_db_connection()
try:
with conn.cursor() as cur:
cur.execute(
"INSERT INTO users (username, email) VALUES (%s, %s) RETURNING id, username, email, created_at",
(username, email)
)
row = cur.fetchone()
conn.commit()
user = {
"id": row[0],
"username": row[1],
"email": row[2],
"created_at": row[3].isoformat()
}
# Warm the cache proactively (cache-aside write-through variant)
self.cache.set(f"user:{user['id']}", user, ttl=600)
self.cache.set(f"user:username:{username.lower()}", user, ttl=600)
return user
except Exception:
conn.rollback()
raise
finally:
self._release_db_connection(conn)
For create_user, we proactively populate the cache (sometimes called cache warming) because we already have the full object. This avoids an immediate cache miss on the first read after creation.
Step 5: Caching Complex Queries and Collections
Beyond single-record lookups, you'll often want to cache list results — like "top 10 products" or "recent articles." These require careful key design and invalidation strategies:
def get_active_users(self, limit: int = 20) -> list[dict]:
"""Cache a list query with a shorter TTL since collections change more frequently."""
cache_key = f"users:active:{limit}"
cached = self.cache.get(cache_key)
if cached is not None:
return cached
conn = self._get_db_connection()
try:
with conn.cursor() as cur:
cur.execute(
"""SELECT id, username, email, created_at
FROM users
ORDER BY created_at DESC
LIMIT %s""",
(limit,)
)
rows = cur.fetchall()
users = [
{
"id": r[0],
"username": r[1],
"email": r[2],
"created_at": r[3].isoformat()
}
for r in rows
]
# Collections change more often — use a shorter TTL
self.cache.set(cache_key, users, ttl=60) # 1 minute
return users
finally:
self._release_db_connection(conn)
When you update a user, you must also invalidate collection caches that include that user. The delete_pattern method helps here:
def invalidate_user_caches(self, user_id: int, username: str):
"""Bulk-invalidate all cache keys related to a user."""
# Delete specific keys
self.cache.delete(f"user:{user_id}")
self.cache.delete(f"user:username:{username.lower()}")
# Delete all collection caches that might contain this user
self.cache.delete_pattern("users:active:*")
self.cache.delete_pattern("users:search:*")
print(f" ✓ Bulk-invalidated all caches for user:{user_id}")
Step 6: Wiring Everything Together
Here is a complete main.py that initializes the Redis client, the cache layer, and the repository, then demonstrates the full read/write cycle:
# main.py
import redis
from cache_layer import CacheLayer
from user_repository import UserRepository
def main():
# Initialize Redis client with connection pooling
redis_client = redis.Redis(
host="localhost",
port=6379,
password="myredispassword",
decode_responses=True,
socket_connect_timeout=2,
socket_timeout=2,
max_connections=20,
retry_on_timeout=True
)
# Verify Redis connectivity
try:
redis_client.ping()
print("✓ Redis connected successfully")
except redis.ConnectionError:
print("⚠ Redis unavailable — application will run without cache")
# In production, you'd fall back gracefully here
# Create the cache layer
cache = CacheLayer(redis_client, default_ttl=300)
# Create the repository
repo = UserRepository(cache)
# ---- DEMONSTRATION ----
print("\n--- First read (cache miss) ---")
user = repo.get_user_by_id(1)
print(f"Result: {user}\n")
print("--- Second read (should be cache hit) ---")
user = repo.get_user_by_id(1)
print(f"Result: {user}\n")
print("--- Update user email (invalidates cache) ---")
repo.update_user_email(1, "alice.new@example.com")
print()
print("--- Read after update (cache miss — fresh data) ---")
user = repo.get_user_by_id(1)
print(f"Result: {user}\n")
print("--- Create a new user (cache warmed) ---")
new_user = repo.create_user("diana", "diana@example.com")
print(f"Created: {new_user}\n")
print("--- Read new user (cache hit — warmed on create) ---")
user = repo.get_user_by_id(new_user["id"])
print(f"Result: {user}\n")
# Cleanup
redis_client.close()
if __name__ == "__main__":
main()
Run it and observe the output — you'll see explicit cache hits and misses logged to the console, confirming the pattern works end-to-end.
Best Practices for a Redis Cache Layer
Building a cache layer is straightforward; building one that behaves predictably in production requires discipline. Here are the practices that separate reliable systems from fragile ones:
- Always set a TTL. Every cached key needs an expiration. Without it, stale data accumulates and Redis memory fills up. Choose TTLs based on how frequently the underlying data changes — 5 minutes for user profiles, 30 seconds for real-time metrics, 1 hour for reference data like country lists
- Cache key naming conventions. Use structured, namespaced keys like
entity:identifier:subtype. Examples:user:42,user:username:alice,product:featured:10. This makes debugging, monitoring, and pattern-based invalidation trivial - Handle Redis failures gracefully. Redis can and will go down. Your cache layer must degrade to database-only operation without throwing errors. The wrapper's try/except blocks above ensure this. Never let a cache failure cascade into an application outage
- Invalidate, don't update. When data changes in PostgreSQL, delete the cache key rather than updating it in-place. Invalidation avoids race conditions where two writers update cache and database in different orders. Let the next read rebuild the cache from the authoritative source
- Use connection pooling. Both Redis (
redis.ConnectionPool) and PostgreSQL (psycopg2.pool) benefit from connection reuse. Creating new connections on every request adds latency and exhausts server resources - Monitor cache hit rates. Track
cache_hits / (cache_hits + cache_misses)as a metric. A hit rate below 70% suggests your TTLs are too short or your cache key design doesn't match access patterns. Above 90% means you're effectively offloading PostgreSQL - Avoid caching large blobs. Redis performs best with values under a few kilobytes. If you're caching large JSON documents or binary data, consider storing only IDs in Redis and fetching full objects from PostgreSQL, or use Redis Streams for large payloads
- Use cache warming selectively. Pre-populating cache on writes (as in
create_user) is useful but not always necessary. For high-throughput write endpoints, skip warming to avoid doubling the Redis write load — the first read can do the backfill naturally - Consider cache stampedes. When a popular key expires, dozens of concurrent requests can all miss and hit PostgreSQL simultaneously. For high-traffic keys, implement a simple lock or use Redis's
SETNXto let only one request rebuild the cache while others wait briefly
Advanced: Cache Stampede Prevention
A cache stampede happens when a heavily-requested key expires and many concurrent requests simultaneously hit the database to rebuild it. Here's a lightweight solution using Redis locks:
import time
def get_user_by_id_safe(self, user_id: int) -> dict | None:
cache_key = f"user:{user_id}"
lock_key = f"lock:user:{user_id}"
# Try cache first
cached = self.cache.get(cache_key)
if cached is not None:
return cached
# Cache miss — try to acquire rebuild lock
acquired_lock = False
try:
# SETNX is atomic — only one caller succeeds
acquired_lock = self.cache.redis.setnx(lock_key, "1")
if acquired_lock:
# Set lock expiry to prevent deadlocks
self.cache.redis.expire(lock_key, 5)
# Rebuild from PostgreSQL
user = self._fetch_and_cache_user(user_id, cache_key)
return user
else:
# Another process is rebuilding — wait briefly and retry cache
for _ in range(10):
time.sleep(0.05) # 50ms
cached = self.cache.get(cache_key)
if cached is not None:
return cached
# Fallback: query PostgreSQL directly after waiting
return self._fetch_and_cache_user(user_id, cache_key)
finally:
if acquired_lock:
self.cache.redis.delete(lock_key)
This pattern ensures only one request pays the database cost when a popular key expires. All other callers either get the rebuilt cache within 500ms or fall back to a direct query.
Conclusion
A Redis cache layer for PostgreSQL is one of the highest-leverage optimizations you can make to a web application. The Cache-Aside pattern is simple to implement, keeps your database as the authoritative source of truth, and gracefully degrades when Redis is unavailable. By following the practices outlined here — structured key namespaces, mandatory TTLs, invalidation-on-write, connection pooling, and stampede protection — you'll build a cache layer that scales reads effortlessly, reduces database costs, and keeps response times consistently fast under load. The complete code in this tutorial forms a solid foundation; adapt the TTLs, key patterns, and error handling to your specific domain, and you'll have a production-grade caching infrastructure in place.