KAHIBARO
Discord Login Register ☰
☰

Redis Integration

Table of Contents

Why Use Redis in the Final Project?

By the time you reach this chapter, your final project should already have:

Redis adds two important capabilities:

  1. Fast, in-memory key-value storage for things like:
    • Caching expensive database queries
    • Session or token blacklists
    • Rate limiting and throttling
    • Storing transient data for background jobs
  2. Lightweight coordination between services, for example:
    • Sharing state between your API and workers
    • Distributed locks to avoid race conditions

In this chapter, you will integrate Redis into your final project in a way that is structured, testable, and production-ready.

Designing Redis Usage in Your Architecture

Before writing code, decide clearly what Redis is responsible for in your project.

Typical uses for a production-style backend:

Use caseExample key patternTypical TTL (time to live)
Caching DB queriesproduct:{id}, user:{id}5–30 minutes
API rate limitingrate:{user_id}:{endpoint}1 minute
Session / token revokeblacklist:{jti} or session:{id}Until token/session expiration
Background job metadatajob:{id}Until job completion + 1 day
Feature flags (optional)feature:{name}No expiration or long TTL

Rule: Redis must never be your only source of truth for critical data.
All important, persistent data must still be stored in PostgreSQL or another durable store.

Decide and document:

Create a small design document in your project repo, for example:
docs/redis-usage.md.

Configuring Redis in the Project

Environment Variables

Add Redis configuration to your environment variables.

Example .env entries:

env 
REDIS_HOST=redis
REDIS_PORT=6379
REDIS_DB=0
REDIS_PASSWORD=
REDIS_USE_TLS=false

In production you might have:

env 
REDIS_URL=redis://default:strong-password@my-redis-host:6379/0

Decide whether you will:

Then be consistent across your codebase.

Configuration Module

Create or extend a central config (for example with Pydantic settings):

python 
from pydantic import BaseSettings, AnyUrl
class Settings(BaseSettings):
    redis_url: AnyUrl | None = None
    redis_host: str = "localhost"
    redis_port: int = 6379
    redis_db: int = 0
    redis_password: str | None = None
    class Config:
        env_file = ".env"
settings = Settings()

A helper to build the URL if REDIS_URL is not given:

python 
from urllib.parse import quote_plus
def get_redis_url_from_settings(settings: Settings) -> str:
    if settings.redis_url:
        return str(settings.redis_url)
    pwd_segment = ""
    if settings.redis_password:
        pwd_segment = f":{quote_plus(settings.redis_password)}@"
    return (
        f"redis://{pwd_segment}"
        f"{settings.redis_host}:{settings.redis_port}/{settings.redis_db}"
    )

Choosing a Redis Client Library

For a modern FastAPI-based backend, a common choice is redis-py with async support.

Install:

bash 
pip install redis[hiredis]

This gives you:

For an async-first backend, it is better to use redis.asyncio.Redis.

Creating a Redis Client and Dependency

Central Redis Client Creation

Create app/core/redis.py (or similar):

python 
from redis.asyncio import Redis
from typing import AsyncGenerator
from app.core.config import settings, get_redis_url_from_settings
_redis_client: Redis | None = None
async def get_redis_client() -> Redis:
    global _redis_client
    if _redis_client is None:
        url = get_redis_url_from_settings(settings)
        _redis_client = Redis.from_url(
            url,
            encoding="utf-8",
            decode_responses=True,  # return strings instead of bytes
        )
    return _redis_client
async def redis_lifespan(app) -> AsyncGenerator[None, None]:
    client = await get_redis_client()
    try:
        # Verify connection at startup
        await client.ping()
        yield
    finally:
        await client.close()

You can hook redis_lifespan into FastAPI’s lifespan handler in your app initialization chapter, so here focus only on the Redis-specific part.

FastAPI Dependency

Create a dependency that can be injected into routes or services:

python 
from fastapi import Depends
from redis.asyncio import Redis
from app.core.redis import get_redis_client
async def get_redis(
    client: Redis = Depends(get_redis_client),
) -> Redis:
    return client

Use this dependency only in the top layers (like routes) and in infrastructure services. Do not pass Redis directly into your domain logic.

Implementing a Caching Layer

Designing a Reusable Cache Service

Create app/services/cache.py:

python 
from typing import Any, Callable, Awaitable
import json
import functools
from redis.asyncio import Redis
class CacheService:
    def __init__(self, redis: Redis, default_ttl_seconds: int = 300):
        self._redis = redis
        self._default_ttl = default_ttl_seconds
    async def get(self, key: str) -> Any | None:
        raw = await self._redis.get(key)
        if raw is None:
            return None
        return json.loads(raw)
    async def set(
        self,
        key: str,
        value: Any,
        ttl_seconds: int | None = None,
    ) -> None:
        ttl = ttl_seconds or self._default_ttl
        await self._redis.set(key, json.dumps(value), ex=ttl)
    async def delete(self, key: str) -> None:
        await self._redis.delete(key)
    async def exists(self, key: str) -> bool:
        return bool(await self._redis.exists(key))

Then a small factory dependency:

python 
from fastapi import Depends
from redis.asyncio import Redis
from app.core.redis import get_redis_client
from app.services.cache import CacheService
def get_cache_service(
    redis: Redis = Depends(get_redis_client),
) -> CacheService:
    return CacheService(redis)

Caching a Database Query

Imagine you have a product service:

python 
class ProductService:
    def __init__(self, repo, cache: CacheService | None = None):
        self._repo = repo
        self._cache = cache
    async def get_product_by_id(self, product_id: int):
        cache_key = f"product:{product_id}"
        # 1. Try from cache
        if self._cache:
            cached = await self._cache.get(cache_key)
            if cached is not None:
                return cached
        # 2. Fallback to DB
        product = await self._repo.get_by_id(product_id)
        if product is None:
            return None
        # 3. Save into cache
        if self._cache:
            await self._cache.set(cache_key, product, ttl_seconds=600)
        return product

Example route:

python 
from fastapi import APIRouter, Depends, HTTPException, status
from app.services.cache import CacheService, get_cache_service
from app.services.products import ProductService, get_product_repo
router = APIRouter()
def get_product_service(
    repo = Depends(get_product_repo),
    cache: CacheService = Depends(get_cache_service),
) -> ProductService:
    return ProductService(repo=repo, cache=cache)
@router.get("/products/{product_id}")
async def read_product(
    product_id: int,
    service: ProductService = Depends(get_product_service),
):
    product = await service.get_product_by_id(product_id)
    if product is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
    return product

Note how the route does not know any Redis details. It just uses a ProductService.

Cache Invalidation

Caching only works correctly if you invalidate or update cached data when the underlying data changes.

For products, whenever you update or delete a product, you should clear the cache for that product.

Example in ProductService:

python 
    async def update_product(self, product_id: int, data) -> dict | None:
        updated = await self._repo.update(product_id, data)
        if updated and self._cache:
            cache_key = f"product:{product_id}"
            await self._cache.delete(cache_key)
        return updated
    async def delete_product(self, product_id: int) -> bool:
        deleted = await self._repo.delete(product_id)
        if deleted and self._cache:
            cache_key = f"product:{product_id}"
            await self._cache.delete(cache_key)
        return deleted

Rule: Every write that affects cached data must either update or invalidate the relevant cache keys.
If you forget this, your API may return stale or inconsistent data.

For list endpoints, for example GET /products?category_id=123, a simple pattern is:

Rate Limiting with Redis

Rate limiting prevents a single client from sending too many requests in a short time.

A simple per-user, per-endpoint limit:

Simple Sliding Window Implementation

Create app/services/rate_limiter.py:

python 
from redis.asyncio import Redis
from fastapi import HTTPException, status
class RateLimiter:
    def __init__(self, redis: Redis, limit: int, window_seconds: int):
        self.redis = redis
        self.limit = limit
        self.window = window_seconds
    async def check(self, key: str) -> None:
        # INCR returns the new value after increment
        current = await self.redis.incr(key)
        if current == 1:
            # First hit, set expiration
            await self.redis.expire(key, self.window)
        if current > self.limit:
            raise HTTPException(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                detail="Too many requests, please try again later.",
            )

Now a FastAPI dependency that uses it:

python 
from fastapi import Depends, Request
from redis.asyncio import Redis
from app.core.redis import get_redis_client
from app.services.rate_limiter import RateLimiter
def rate_limiter_dependency(
    limit: int,
    window_seconds: int,
):
    async def dependency(
        request: Request,
        redis: Redis = Depends(get_redis_client),
    ):
        user_id = request.state.user_id if hasattr(request.state, "user_id") else "anonymous"
        endpoint = request.url.path
        key = f"rate:{user_id}:{endpoint}"
        limiter = RateLimiter(redis, limit, window_seconds)
        await limiter.check(key)
    return dependency

Use it on a route:

python 
from fastapi import Depends
from app.services.rate_limiter_dependency import rate_limiter_dependency
@router.get("/orders")
async def list_orders(
    _=Depends(rate_limiter_dependency(limit=60, window_seconds=60)),
):
    ...

This limits each user to 60 requests per minute to /orders.

Token Blacklisting or Session Storage

If you implement JWT access and refresh tokens, you may want to:

Storing Sessions

A simple session model:

Example repository:

python 
import json
from datetime import timedelta
from redis.asyncio import Redis
class SessionStore:
    def __init__(self, redis: Redis):
        self._redis = redis
    async def create_session(
        self,
        session_id: str,
        data: dict,
        ttl: timedelta,
    ) -> None:
        key = f"session:{session_id}"
        await self._redis.set(
            key,
            json.dumps(data),
            ex=int(ttl.total_seconds()),
        )
    async def get_session(self, session_id: str) -> dict | None:
        key = f"session:{session_id}"
        raw = await self._redis.get(key)
        if raw is None:
            return None
        return json.loads(raw)
    async def delete_session(self, session_id: str) -> None:
        key = f"session:{session_id}"
        await self._redis.delete(key)

Your authentication layer can then:

Blacklisting JWTs

JWTs are stateless, so to support logout you can:

Example:

python 
class TokenBlacklist:
    def __init__(self, redis: Redis):
        self._redis = redis
    async def add(self, jti: str, ttl_seconds: int) -> None:
        key = f"blacklist:{jti}"
        await self._redis.set(key, "1", ex=ttl_seconds)
    async def is_blacklisted(self, jti: str) -> bool:
        key = f"blacklist:{jti}"
        return bool(await self._redis.exists(key))

Your auth dependency can then:

Using Redis from Background Workers

Your background worker processes (for example Celery or a custom worker) can also use Redis.

Typical roles:

Key point: use the same configuration approach.

In your worker process:

Example in a worker module:

python 
from redis.asyncio import Redis
from app.core.config import settings, get_redis_url_from_settings
def create_redis_for_worker() -> Redis:
    url = get_redis_url_from_settings(settings)
    return Redis.from_url(
        url,
        encoding="utf-8",
        decode_responses=True,
    )

Use this function in your task handlers to get a Redis client.

Handling Redis Failures Gracefully

In production, Redis may be temporarily unavailable. You must decide how the system behaves when Redis fails.

Typical strategy:

Example: Safe Cache Wrapper

Update your CacheService methods to catch Redis errors:

python 
import logging
from redis.exceptions import RedisError
logger = logging.getLogger(__name__)
class CacheService:
    ...
    async def get(self, key: str):
        try:
            raw = await self._redis.get(key)
        except RedisError:
            logger.exception("Redis GET failed")
            return None
        if raw is None:
            return None
        return json.loads(raw)
    async def set(self, key: str, value: Any, ttl_seconds: int | None = None):
        try:
            ttl = ttl_seconds or self._default_ttl
            await self._redis.set(key, json.dumps(value), ex=ttl)
        except RedisError:
            logger.exception("Redis SET failed, ignoring")

For rate limiting:

python 
from redis.exceptions import RedisError
class RateLimiter:
    ...
    async def check(self, key: str) -> None:
        try:
            current = await self.redis.incr(key)
            if current == 1:
                await self.redis.expire(key, self.window)
        except RedisError:
            # Fail-open: if Redis is down, skip limiting
            return
        if current > self.limit:
            ...

Rule: Decide per feature whether Redis failures should:

  • Fail-open, allow the request (caches, most rate limits), or
  • Fail-closed, deny the request (security-sensitive session checks)
    Document this behavior.

Testing Redis Integration

To keep tests fast and deterministic, avoid using a real Redis instance for unit tests if possible.

Common patterns:

  1. Use a mock Redis client for unit tests of higher-level services
  2. Use a real Redis (with Docker or local) for integration tests

Mocking Redis for Unit Tests

Create a simple in-memory fake:

python 
class FakeRedis:
    def __init__(self):
        self.store = {}
    async def get(self, key: str):
        return self.store.get(key)
    async def set(self, key: str, value, ex=None):
        self.store[key] = value
    async def delete(self, key: str):
        self.store.pop(key, None)
    async def incr(self, key: str):
        value = int(self.store.get(key, "0"))
        value += 1
        self.store[key] = str(value)
        return value
    async def expire(self, key: str, seconds: int):
        # For unit tests, you can ignore TTL
        return True
    async def exists(self, key: str):
        return 1 if key in self.store else 0

Use this fake in tests:

python 
import pytest
from app.services.cache import CacheService
from tests.fakes import FakeRedis
@pytest.mark.asyncio
async def test_cache_service_get_set():
    redis = FakeRedis()
    cache = CacheService(redis)
    assert await cache.get("foo") is None
    await cache.set("foo", {"a": 1})
    assert await cache.get("foo") == {"a": 1}

Integration Tests with a Real Redis

If you already use Docker Compose, add a Redis service, then in tests:

Example fixture:

python 
import pytest
from redis.asyncio import Redis
from app.core.config import settings, get_redis_url_from_settings
@pytest.fixture
async def redis_client():
    url = get_redis_url_from_settings(settings)
    client = Redis.from_url(url, encoding="utf-8", decode_responses=True)
    await client.flushdb()
    try:
        yield client
    finally:
        await client.close()

Operational Considerations

For production, also think about:

Summary

In the final project, Redis should:

With this integration in place, your final project closely resembles a real production backend that combines PostgreSQL for durable data and Redis for fast, transient operations.

Views: 6

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 3096 / 24h
Impressum / Imprint
Datenschutz / Data Policy