redis-pattern

name

redis-pattern

description

Generate Redis patterns for Python async applications using redis-py and arq. Use when implementing: (1) Caching with TTL and invalidation, (2) Pub/Sub for real-time events, (3) arq task queues for background jobs, (4) Distributed locks, (5) Rate limiting. Generates production-ready async code with proper connection pooling, error handling, and typing. NOT for Celery, RQ, or synchronous Redis patterns.

Redis Pattern Generator

Generate async Redis patterns for FastAPI microservices using redis-py and arq.

Quick Reference

Pattern	Command	Output
Cache	/redis-pattern cache	Async cache with TTL, invalidation
Pub/Sub	/redis-pattern pubsub	Event publisher/subscriber
Task Queue	/redis-pattern task-queue	arq worker and task definitions
Lock	/redis-pattern lock	Distributed lock context manager
Rate Limit	/redis-pattern rate-limit	Sliding window rate limiter
All	/redis-pattern	Full Redis infrastructure setup

Pattern Selection

Choose pattern based on use case:

Need to...                          Pattern
─────────────────────────────────────────────
Cache expensive operations       → cache
Send real-time events            → pubsub
Run background jobs              → task-queue
Prevent race conditions          → lock
Limit request frequency          → rate-limit
Set up full Redis infrastructure → (no args)

Implementation

1. Connection Setup (Always Required)

Generate in core/redis.py:

from redis.asyncio import Redis, ConnectionPool
from contextlib import asynccontextmanager
from typing import AsyncGenerator
import os

_pool: ConnectionPool | None = None

def get_redis_url() -> str:
    return os.getenv("REDIS_URL", "redis://localhost:6379/0")

async def init_redis_pool() -> None:
    global _pool
    _pool = ConnectionPool.from_url(
        get_redis_url(),
        max_connections=20,
        decode_responses=True,
    )

async def close_redis_pool() -> None:
    global _pool
    if _pool:
        await _pool.disconnect()
        _pool = None

@asynccontextmanager
async def get_redis() -> AsyncGenerator[Redis, None]:
    if not _pool:
        raise RuntimeError("Redis pool not initialized")
    client = Redis(connection_pool=_pool)
    try:
        yield client
    finally:
        await client.aclose()

# For dependency injection in FastAPI
async def redis_dependency() -> AsyncGenerator[Redis, None]:
    async with get_redis() as redis:
        yield redis

2. Pattern Implementation

See references for detailed patterns:

• Caching: See references/cache.md
• Pub/Sub: See references/pubsub.md
• Task Queue (arq): See references/task-queue.md
• Distributed Lock: See references/lock.md
• Rate Limiting: See references/rate-limit.md

3. FastAPI Integration

from fastapi import FastAPI
from contextlib import asynccontextmanager
from core.redis import init_redis_pool, close_redis_pool

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_redis_pool()
    yield
    await close_redis_pool()

app = FastAPI(lifespan=lifespan)

File Structure

service/
├── core/
│   └── redis.py          # Connection pool (always)
├── cache/
│   └── decorators.py     # @cached decorator (if cache)
├── events/
│   ├── publisher.py      # Event publishing (if pubsub)
│   └── subscriber.py     # Event handlers (if pubsub)
├── tasks/
│   ├── worker.py         # arq worker config (if task-queue)
│   └── definitions.py    # Task functions (if task-queue)
└── utils/
    ├── lock.py           # Distributed lock (if lock)
    └── rate_limit.py     # Rate limiter (if rate-limit)

Dependencies

Add to pyproject.toml:

[project]
dependencies = [
    "redis>=5.0.0",        # Async Redis client
    "arq>=0.26.0",         # Task queue (if needed)
    "hiredis>=3.0.0",      # C parser for performance
]

Environment Variables

REDIS_URL=redis://localhost:6379/0
ARQ_QUEUE_NAME=omnichat:tasks        # For arq

Testing

import pytest
from unittest.mock import AsyncMock
from redis.asyncio import Redis

@pytest.fixture
def mock_redis():
    redis = AsyncMock(spec=Redis)
    redis.get.return_value = None
    redis.set.return_value = True
    return redis