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

> Learn to configure Redis for cache queue and rate limiting in FastAPI using benavlabs fastapi boilerplate. Simplify your app performance with Pydantic settings and lifespan events.

- Repository: [Benav Labs/fastapi-boilerplate](https://github.com/benavlabs/fastapi-boilerplate)
- Tags: how-to-guide
- Published: 2026-02-26

---

**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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/config.py)

Redis configuration is centralized in [`src/app/core/config.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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:

```python

# 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:

```python

# 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:

```python

# 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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/setup.py)

The [`src/app/core/setup.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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

```python

# 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:

```python

# 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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/utils/cache.py) module provides a `@cache` decorator that utilizes the global cache client initialized during lifespan startup.

### Cache Decorator Usage

```python
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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/api/v1/tasks.py) endpoint demonstrates how to enqueue background tasks:

```python
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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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:

```python

# 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:

```python
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

```dotenv

# 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:

```dotenv
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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/config.py).

- **Connection pools** are created during FastAPI lifespan startup in [`src/app/core/setup.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/utils/queue.py) and are enqueued in [`src/app/api/v1/tasks.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/api/v1/tasks.py) for asynchronous processing.

- **Rate limiting** implements fixed-window counters in [`src/app/core/utils/rate_limit.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/config.py). The lifespan factory in [`src/app/core/setup.py`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/core/utils/cache.py) accepts an `expiration` parameter in seconds. Pass this when applying the decorator to your endpoint:

```python
@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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/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`](https://github.com/benavlabs/fastapi-boilerplate/blob/main/src/app/api/v1/health.py) monitor Redis connectivity to detect failures before they impact users.