How to Configure Redis for Cache, Queue, and Rate Limiting in FastAPI

The benavlabs/fastapi-boilerplate provides three independent Redis integrations for caching, background job queuing via ARQ, and rate limiting, all configured through Pydantic settings and managed via FastAPI lifespan events.

The benavlabs/fastapi-boilerplate implements a production-ready Redis stack that separates concerns across three distinct connection pools. This architecture allows you to configure Redis for cache, queue, and rate limiting independently while maintaining clean separation of concerns and optimal resource utilization.

Architecture Overview

The boilerplate implements three distinct Redis concerns, each with dedicated settings, connection pools, and utility modules:

Concern	Settings Class	Connection Pool	Primary Usage
Cache	RedisCacheSettings	create_redis_cache_pool()	Response caching via @cache decorator
Queue	RedisQueueSettings	create_redis_queue_pool()	ARQ background job processing
Rate Limiter	RedisRateLimiterSettings	create_redis_rate_limit_pool()	Fixed-window rate limiting

All three pools are initialized during the FastAPI lifespan startup and gracefully closed during shutdown.

Centralized Configuration in config.py

Redis configuration is centralized in src/app/core/config.py using Pydantic settings classes that read from environment variables.

Cache Settings

The RedisCacheSettings class constructs a Redis URL from discrete environment variables:

# src/app/core/config.py

class RedisCacheSettings(BaseSettings):
    REDIS_CACHE_HOST: str = "localhost"
    REDIS_CACHE_PORT: int = 6379
    REDIS_CACHE_PASSWORD: str | None = None
    REDIS_CACHE_DB: int = 0
    
    @property
    def REDIS_CACHE_URL(self) -> str:
        auth = f":{self.REDIS_CACHE_PASSWORD}@" if self.REDIS_CACHE_PASSWORD else ""
        return f"redis://{auth}{self.REDIS_CACHE_HOST}:{self.REDIS_CACHE_PORT}/{self.REDIS_CACHE_DB}"

Queue Settings

The RedisQueueSettings class exposes host and port separately because the ARQ client expects a RedisSettings object rather than a URL:

# src/app/core/config.py

class RedisQueueSettings(BaseSettings):
    REDIS_QUEUE_HOST: str = "localhost"
    REDIS_QUEUE_PORT: int = 6379

Rate Limiter Settings

The RedisRateLimiterSettings follows the same URL pattern as the cache but uses a separate database (default DB 2) to isolate rate limit counters:

# src/app/core/config.py

class RedisRateLimiterSettings(BaseSettings):
    REDIS_RATE_LIMIT_HOST: str = "localhost"
    REDIS_RATE_LIMIT_PORT: int = 6379
    REDIS_RATE_LIMIT_DB: int = 2
    
    @property
    def REDIS_RATE_LIMIT_URL(self) -> str:
        return f"redis://{self.REDIS_RATE_LIMIT_HOST}:{self.REDIS_RATE_LIMIT_PORT}/{self.REDIS_RATE_LIMIT_DB}"

Lifespan-Based Pool Management in setup.py

The src/app/core/setup.py module manages Redis connection lifecycles using FastAPI's lifespan context manager. During application startup, the lifespan_factory creates pools based on which settings classes are included in your Settings configuration.

Pool Creation Functions

# src/app/core/setup.py

async def create_redis_cache_pool():
    """Creates Redis connection pool for caching."""
    from app.core.utils.cache import cache
    cache.client = await redis.from_url(
        settings.REDIS_CACHE_URL,
        encoding="utf-8",
        decode_responses=True,
        max_connections=20
    )

async def create_redis_queue_pool():
    """Creates ARQ Redis pool for background jobs."""
    from app.core.utils.queue import queue
    from arq import create_pool
    from arq.connections import RedisSettings
    
    redis_settings = RedisSettings(
        host=settings.REDIS_QUEUE_HOST,
        port=settings.REDIS_QUEUE_PORT
    )
    queue.pool = await create_pool(redis_settings)

async def create_redis_rate_limit_pool():
    """Creates Redis connection pool for rate limiting."""
    from app.core.utils.rate_limit import rate_limiter
    rate_limiter.client = await redis.from_url(
        settings.REDIS_RATE_LIMIT_URL,
        encoding="utf-8",
        decode_responses=True
    )

Lifespan Integration

The lifespan factory conditionally initializes pools based on the settings instance type:

# src/app/core/setup.py

@asynccontextmanager
async def lifespan_factory(app: FastAPI):
    # Initialize based on available settings

    if isinstance(settings, RedisCacheSettings):
        await create_redis_cache_pool()
    if isinstance(settings, RedisQueueSettings):
        await create_redis_queue_pool()
    if isinstance(settings, RedisRateLimiterSettings):
        await create_redis_rate_limit_pool()
    
    yield
    
    # Cleanup

    if isinstance(settings, RedisCacheSettings):
        await close_redis_cache_pool()
    if isinstance(settings, RedisQueueSettings):
        await close_redis_queue_pool()
    if isinstance(settings, RedisRateLimiterSettings):
        await close_redis_rate_limit_pool()

Implementing Redis Cache

The src/app/core/utils/cache.py module provides a @cache decorator that utilizes the global cache client initialized during lifespan startup.

Cache Decorator Usage

from fastapi import APIRouter, Request
from app.core.utils.cache import cache

router = APIRouter()

@router.get("/users/{user_id}")
@cache(key_prefix="user_profile", expiration=600)
async def get_user_profile(request: Request, user_id: int):
    """
    Caches GET responses for 10 minutes.
    Non-GET requests automatically invalidate the cache key.
    """
    return await user_service.fetch_profile(user_id)

Cache Invalidation

The decorator automatically handles cache invalidation on POST, PUT, PATCH, and DELETE requests by deleting the associated cache key and any extra invalidation patterns configured.

Configuring Redis Queue for Background Jobs

The src/app/core/utils/queue.py module wraps an ARQ Redis pool for asynchronous job processing.

Enqueueing Jobs

The src/app/api/v1/tasks.py endpoint demonstrates how to enqueue background tasks:

from fastapi import APIRouter, HTTPException
from app.core.utils import queue

router = APIRouter()

@router.post("/tasks/notify")
async def trigger_notification(message: str):
    """
    Enqueues a background notification task.
    Returns 503 if Redis queue is not initialized.
    """
    if queue.pool is None:
        raise HTTPException(status_code=503, detail="Queue not ready")
    
    job = await queue.pool.enqueue_job("sample_background_task", message)
    return {"job_id": job.job_id, "status": "queued"}

Worker Configuration

Jobs are defined in worker modules (e.g., src/app/workers/sample_background_task.py) and executed by separate ARQ worker processes that connect to the same Redis instance configured in RedisQueueSettings.

Setting Up Redis Rate Limiting

The src/app/core/utils/rate_limit.py implements a fixed-window rate limiting algorithm using Redis counters.

Rate Limiter Implementation

The RateLimiter class uses Redis INCR and EXPIRE commands to implement fixed-window counters:

# src/app/core/utils/rate_limit.py

class RateLimiter:
    def __init__(self):
        self.client = None
    
    async def is_rate_limited(
        self, 
        db, 
        user_id: int, 
        path: str, 
        limit: int, 
        period: int
    ) -> bool:
        """
        Fixed-window rate limiting.
        Returns True if request should be blocked.
        """
        window_start = int(time.time()) // period
        key = f"ratelimit:{user_id}:{path}:{window_start}"
        
        current_count = await self.client.incr(key)
        
        # Set expiry on first request in window

        if current_count == 1:
            await self.client.expire(key, period)
        
        return current_count > limit

FastAPI Dependency Integration

Create a reusable dependency that applies rate limiting to specific endpoints:

from fastapi import Depends, Request, HTTPException
from app.core.utils.rate_limit import rate_limiter
from app.core.schemas.rate_limit import sanitize_path

async def rate_limiter_dependency(
    request: Request,
    user_id: int = Depends(get_current_user_id),
):
    """
    Dependency that enforces rate limits per user per endpoint.
    """
    path = sanitize_path(request.url.path)
    limited = await rate_limiter.is_rate_limited(
        db=None,
        user_id=user_id,
        path=path,
        limit=100,      # requests per period

        period=3600,    # 1 hour window

    )
    if limited:
        raise HTTPException(status_code=429, detail="Too many requests")

# Apply to router

router = APIRouter(dependencies=[Depends(rate_limiter_dependency)])

Environment Variable Configuration

All Redis connections are configured via environment variables defined in .env files. The boilerplate provides examples in scripts/local_with_uvicorn/.env.example and scripts/production_with_nginx/.env.example.

Complete Environment Configuration

# Redis Cache Configuration

REDIS_CACHE_HOST=localhost
REDIS_CACHE_PORT=6379
REDIS_CACHE_PASSWORD=supersecret
REDIS_CACHE_DB=0

# Redis Queue (ARQ) Configuration

REDIS_QUEUE_HOST=localhost
REDIS_QUEUE_PORT=6379

# Redis Rate Limiter Configuration

REDIS_RATE_LIMIT_HOST=localhost
REDIS_RATE_LIMIT_PORT=6379
REDIS_RATE_LIMIT_DB=2

Docker Compose Considerations

When running with Docker Compose, use service names as hosts:

REDIS_CACHE_HOST=redis-cache
REDIS_QUEUE_HOST=redis-queue
REDIS_RATE_LIMIT_HOST=redis-rate

Production Best Practices

When deploying the FastAPI boilerplate with Redis in production, consider these optimizations:

• Database Isolation: The boilerplate defaults to DB 0 for cache, DB 0 for queue (ARQ uses its own key prefixes), and DB 2 for rate limiting. Consider dedicated Redis instances for high-traffic applications to prevent noisy neighbor effects.

• Connection Pool Sizing: Adjust max_connections in create_redis_cache_pool() (default 20) based on your traffic patterns. Monitor connected_clients in Redis INFO to optimize pool sizes.

• Memory Policies: Configure maxmemory-policy volatile-lru on the rate limiter Redis instance to ensure expired counter keys are evicted first, preventing memory exhaustion.

• Health Monitoring: The health check endpoint in src/app/api/v1/health.py verifies Redis connectivity. Ensure your load balancer or orchestrator polls this endpoint to detect connection failures early.

• Security: Use Redis ACLs or password authentication (REDIS_CACHE_PASSWORD) and ensure Redis instances are not exposed to public networks. Consider TLS connections for sensitive data.

Summary

• The benavlabs/fastapi-boilerplate implements three isolated Redis integrations via RedisCacheSettings, RedisQueueSettings, and RedisRateLimiterSettings in src/app/core/config.py.

• Connection pools are created during FastAPI lifespan startup in src/app/core/setup.py through create_redis_cache_pool(), create_redis_queue_pool(), and create_redis_rate_limit_pool().

• Caching uses the @cache decorator from src/app/core/utils/cache.py to store GET responses with automatic invalidation on mutations.

• Background jobs utilize ARQ via src/app/core/utils/queue.py and are enqueued in src/app/api/v1/tasks.py for asynchronous processing.

• Rate limiting implements fixed-window counters in src/app/core/utils/rate_limit.py and integrates as a FastAPI dependency to return HTTP 429 when limits are exceeded.

• All configurations are environment-driven via .env files, supporting separate hosts, ports, and databases for each Redis concern.

Frequently Asked Questions

How do I disable one of the Redis integrations if I don't need it?

Remove the corresponding settings class from your Settings inheritance chain in src/app/core/config.py. The lifespan factory in src/app/core/setup.py uses isinstance checks to conditionally initialize only the pools for which settings are present. For example, to disable rate limiting, ensure your Settings class does not inherit from RedisRateLimiterSettings.

Can I use the same Redis instance for all three concerns?

Yes, but it is not recommended for production workloads. You can point REDIS_CACHE_HOST, REDIS_QUEUE_HOST, and REDIS_RATE_LIMIT_HOST to the same server. However, use different database numbers (e.g., DB 0 for cache, DB 1 for queue, DB 2 for rate limiting) to prevent key collisions. For high-traffic applications, deploy separate Redis instances to avoid resource contention.

How do I adjust the cache expiration time dynamically?

The @cache decorator in src/app/core/utils/cache.py accepts an expiration parameter in seconds. Pass this when applying the decorator to your endpoint:

@cache(key_prefix="dynamic_data", expiration=300)  # 5 minutes

async def get_dynamic_data():
    ...

For dynamic expiration based on business logic, you would need to modify the cache utility to accept a callable or check the decorator source in src/app/core/utils/cache.py to implement custom expiration logic.

What happens if Redis becomes unavailable during runtime?

Each utility handles unavailability differently. The queue utility in src/app/api/v1/tasks.py explicitly checks if queue.pool is None and raises an HTTP 503 error if the pool is not initialized. The cache decorator typically fails silently or allows the request to proceed to the database if the Redis client is disconnected, depending on the implementation in src/app/core/utils/cache.py. The rate limiter will likely raise a connection error if Redis is unavailable unless wrapped in exception handling. For production deployments, ensure your health checks in src/app/api/v1/health.py monitor Redis connectivity to detect failures before they impact users.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}