Build a URL Shortener from Scratch — Part 4: Redis Caching for Fast Redirects

# app/core/cache.py
import redis
from app.config import settings

redis_client = redis.Redis.from_url(
    settings.redis_url,
    decode_responses=True,
    socket_connect_timeout=5,
    socket_timeout=2
)

CACHE_TTL = 60 * 60 * 24 * 7  # 1 week

def cache_url(short_code: str, original_url: str):
    redis_client.setex(f"url:{short_code}", CACHE_TTL, original_url)

def get_cached_url(short_code: str) -> str | None:
    return redis_client.get(f"url:{short_code}")

def invalidate_cache(short_code: str):
    redis_client.delete(f"url:{short_code}")

# app/api/routes.py (updated redirect)
from app.core.cache import get_cached_url, cache_url

@router.get("/{code}")
def redirect_to_url(code: str, db: Session = Depends(get_db)):
    # 1. Check Redis cache first
    cached = get_cached_url(code)
    if cached:
        # Increment click count asynchronously
        increment_click_count.delay(code)  # Celery/background task
        return RedirectResponse(url=cached, status_code=307)

    # 2. Cache miss — query PostgreSQL
    service = ShortenerService(db)
    url = service.get_by_code(code)
    if not url:
        raise HTTPException(status_code=404, detail="Short URL not found")

    # 3. Check expiration
    from datetime import datetime, timezone
    if url.expires_at and url.expires_at < datetime.now(timezone.utc):
        raise HTTPException(status_code=410, detail="Short URL has expired")

    # 4. Cache for next time
    cache_url(code, url.original_url)

    # 5. Increment click counter
    url.click_count += 1
    db.commit()

    return RedirectResponse(url=url.original_url, status_code=307)

# In ShortenerService.create() — after commit
cache_url(url.short_code, url.original_url)

# app/services/analytics.py
import redis as redis_lib
from app.core.cache import redis_client

def increment_click_count_background(short_code: str):
    """Increment click count in Redis, batch-write to PostgreSQL later."""
    redis_client.hincrby("clicks", short_code, 1)

def flush_clicks_to_db(db):
    """Periodic task: flush Redis click counts to PostgreSQL."""
    clicks = redis_client.hgetall("clicks")
    for short_code, count in clicks.items():
        url = db.query(URL).filter(URL.short_code == short_code).first()
        if url:
            url.click_count += int(count)
    db.commit()
    redis_client.delete("clicks")

def update_url(short_code: str, new_url: str, db: Session):
    url = db.query(URL).filter(URL.short_code == short_code).first()
    url.original_url = new_url
    db.commit()
    invalidate_cache(short_code)  # Remove stale cache entry
    cache_url(short_code, new_url)  # Repopulate with new value

# Create a URL
curl -X POST http://localhost:8000/shorten \
  -d '{"url": "https://example.com"}'

# First request (cache miss, PostgreSQL)
curl -w "\nTime: %{time_total}s\n" -o /dev/null -s http://localhost:8000/q0Z
# Time: 0.025s

# Second request (cache hit, Redis)
curl -w "\nTime: %{time_total}s\n" -o /dev/null -s http://localhost:8000/q0Z
# Time: 0.003s — 8x faster!

# Verify cache exists in Redis
redis-cli GET "url:q0Z"
# "https://example.com"