Rate limiting, caching, and performance tuning

Rate Limiting

1. Redis-Based Rate Limiting

Sliding Window Rate Limiter (Most Effective):

python

# app/core/rate_limiting.py
import time
from redis.asyncio import Redis
from fastapi import HTTPException, Request

class RateLimiter:
    def __init__(self, redis: Redis, prefix: str = "rate_limit"):
        self.redis = redis
        self.prefix = prefix
    
    async def is_rate_limited(
        self, 
        identifier: str, 
        max_requests: int, 
        window_seconds: int
    ) -> bool:
        """
        Sliding window rate limiting using sorted sets
        """
        key = f"{self.prefix}:{identifier}"
        now = int(time.time())
        window_start = now - window_seconds
        
        # Remove expired requests
        await self.redis.zremrangebyscore(key, 0, window_start)
        
        # Count requests in current window
        current_count = await self.redis.zcard(key)
        
        if current_count >= max_requests:
            return True
        
        # Add current request
        await self.redis.zadd(key, {str(now): now})
        await self.redis.expire(key, window_seconds)
        
        return False

# Dependency for FastAPI
async def rate_limit_dependency(
    request: Request,
    redis: Redis = Depends(get_redis),
    max_requests: int = 100,
    window_seconds: int = 3600
):
    # Create identifier (IP, user ID, API key, etc.)
    identifier = f"ip:{request.client.host}"
    
    limiter = RateLimiter(redis)
    if await limiter.is_rate_limited(identifier, max_requests, window_seconds):
        raise HTTPException(
            status_code=429, 
            detail="Rate limit exceeded"
        )

# Usage in routes
@app.get("/api/data", dependencies=[Depends(rate_limit_dependency)])
async def get_data():
    return {"data": "protected"}

# Advanced: User-based rate limiting
async def user_rate_limit(user_id: str, plan: str = "free"):
    limits = {
        "free": {"requests": 100, "window": 3600},
        "premium": {"requests": 1000, "window": 3600},
        "enterprise": {"requests": 10000, "window": 3600}
    }
    return limits.get(plan, limits["free"])

2. Token Bucket Algorithm

python

class TokenBucketRateLimiter:
    def __init__(self, redis: Redis):
        self.redis = redis
    
    async def consume(
        self, 
        key: str, 
        capacity: int, 
        refill_rate: float  # tokens per second
    ) -> bool:
        """
        Token bucket algorithm implementation
        """
        now = time.time()
        bucket_key = f"token_bucket:{key}"
        
        # Use Lua script for atomic operations
        lua_script = """
        local key = KEYS[1]
        local now = tonumber(ARGV[1])
        local capacity = tonumber(ARGV[2])
        local refill_rate = tonumber(ARGV[3])
        
        local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
        local tokens = capacity
        local last_refill = now
        
        if bucket[1] then
            tokens = tonumber(bucket[1])
            last_refill = tonumber(bucket[2])
        end
        
        -- Calculate refill
        local time_passed = now - last_refill
        local new_tokens = time_passed * refill_rate
        tokens = math.min(capacity, tokens + new_tokens)
        
        if tokens < 1 then
            return 0
        end
        
        tokens = tokens - 1
        last_refill = now
        
        redis.call('HMSET', key, 'tokens', tokens, 'last_refill', last_refill)
        redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) * 2)
        
        return 1
        """
        
        result = await self.redis.eval(
            lua_script, 
            1,  # number of keys
            bucket_key, 
            now, 
            capacity, 
            refill_rate
        )
        
        return bool(result)

3. Advanced Rate Limiting Strategies

python

# app/middleware/rate_limiting.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, redis: Redis):
        super().__init__(app)
        self.redis = redis
        self.limiter = RateLimiter(redis)
    
    async def dispatch(self, request: Request, call_next):
        # Skip rate limiting for certain paths
        if request.url.path in ["/docs", "/redoc", "/health"]:
            return await call_next(request)
        
        # Different limits for different endpoints
        endpoint_limits = {
            "/api/auth/login": (10, 3600),  # 10 requests per hour
            "/api/users": (100, 3600),      # 100 requests per hour  
            "/api/data": (1000, 3600),      # 1000 requests per hour
        }
        
        limit_config = endpoint_limits.get(
            request.url.path, 
            (100, 3600)  # default
        )
        
        identifier = self._get_identifier(request)
        
        if await self.limiter.is_rate_limited(
            identifier, limit_config[0], limit_config[1]
        ):
            return JSONResponse(
                status_code=429,
                content={"detail": "Rate limit exceeded"}
            )
        
        response = await call_next(request)
        
        # Add rate limit headers
        response.headers["X-RateLimit-Limit"] = str(limit_config[0])
        response.headers["X-RateLimit-Remaining"] = "..."  # Calculate remaining
        
        return response
    
    def _get_identifier(self, request: Request) -> str:
        """Get rate limit identifier from API key, user, or IP"""
        # Check for API key in header
        api_key = request.headers.get("X-API-Key")
        if api_key:
            return f"api_key:{api_key}"
        
        # Check for authenticated user
        user = request.state.user if hasattr(request.state, 'user') else None
        if user:
            return f"user:{user.id}"
        
        # Fall back to IP
        return f"ip:{request.client.host}"

Caching Strategies

1. Multi-Layer Caching Architecture

python

# app/core/caching.py
from typing import Any, Optional, Union
import json
import pickle
from redis.asyncio import Redis

class CacheManager:
    def __init__(self, redis: Redis):
        self.redis = redis
        self.local_cache = {}  # In-memory cache for very hot data
        self.local_ttl = 30  # seconds
    
    async def get(self, key: str) -> Optional[Any]:
        # Try local cache first
        if key in self.local_cache:
            data, timestamp = self.local_cache[key]
            if time.time() - timestamp < self.local_ttl:
                return data
        
        # Try Redis
        redis_data = await self.redis.get(key)
        if redis_data:
            try:
                data = json.loads(redis_data)
                # Update local cache
                self.local_cache[key] = (data, time.time())
                return data
            except json.JSONDecodeError:
                # Handle pickled data
                data = pickle.loads(redis_data)
                self.local_cache[key] = (data, time.time())
                return data
        
        return None
    
    async def set(
        self, 
        key: str, 
        value: Any, 
        expire: int = 3600,
        use_pickle: bool = False
    ):
        # Update local cache
        self.local_cache[key] = (value, time.time())
        
        # Update Redis
        if use_pickle:
            serialized = pickle.dumps(value)
        else:
            serialized = json.dumps(value)
        
        await self.redis.setex(key, expire, serialized)
    
    async def delete(self, key: str):
        # Remove from both caches
        self.local_cache.pop(key, None)
        await self.redis.delete(key)
    
    async def get_or_set(
        self, 
        key: str, 
        fetch_func: callable, 
        expire: int = 3600,
        *args, **kwargs
    ) -> Any:
        """Cache-aside pattern"""
        cached = await self.get(key)
        if cached is not None:
            return cached
        
        # Fetch fresh data
        data = await fetch_func(*args, **kwargs)
        await self.set(key, data, expire)
        return data

# Advanced cache key generation
def generate_cache_key(
    prefix: str, 
    **kwargs
) -> str:
    """Generate consistent cache keys"""
    key_parts = [prefix]
    for k, v in sorted(kwargs.items()):
        if v is not None:
            key_parts.append(f"{k}:{v}")
    return ":".join(key_parts)

2. Database Query Caching

python

# app/core/database_cache.py
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

class CachedDatabaseSession:
    def __init__(self, session: AsyncSession, cache: CacheManager):
        self.session = session
        self.cache = cache
    
    async def execute_cached_query(
        self, 
        query: str,
        params: dict = None,
        expire: int = 300,  # 5 minutes
        force_refresh: bool = False
    ):
        cache_key = generate_cache_key(
            "db_query", 
            query=query, 
            params=json.dumps(params, sort_keys=True)
        )
        
        if not force_refresh:
            cached = await self.cache.get(cache_key)
            if cached is not None:
                return cached
        
        # Execute fresh query
        result = await self.session.execute(text(query), params or {})
        data = [dict(row) for row in result.mappings()]
        
        await self.cache.set(cache_key, data, expire)
        return data

# Usage example
async def get_user_orders_cached(
    user_id: int, 
    db: CachedDatabaseSession,
    cache: CacheManager
):
    query = """
    SELECT o.*, COUNT(oi.id) as item_count 
    FROM orders o 
    LEFT JOIN order_items oi ON o.id = oi.order_id 
    WHERE o.user_id = :user_id 
    GROUP BY o.id
    """
    
    return await db.execute_cached_query(
        query=query,
        params={"user_id": user_id},
        expire=600  # 10 minutes
    )

3. Cache Invalidation Strategies

python

# app/core/cache_invalidation.py
class CacheInvalidator:
    def __init__(self, redis: Redis, cache: CacheManager):
        self.redis = redis
        self.cache = cache
    
    async def invalidate_user_data(self, user_id: int):
        """Invalidate all cache entries for a user"""
        pattern = f"*:user_id:{user_id}*"
        keys = await self.redis.keys(pattern)
        if keys:
            await self.redis.delete(*keys)
        
        # Also clear local cache entries
        local_keys_to_remove = [
            k for k in self.cache.local_cache.keys() 
            if f"user_id:{user_id}" in k
        ]
        for key in local_keys_to_remove:
            self.cache.local_cache.pop(key, None)
    
    async def invalidate_pattern(self, pattern: str):
        """Invalidate keys matching pattern"""
        keys = await self.redis.keys(pattern)
        if keys:
            await self.redis.delete(*keys)
    
    async def setup_cache_invalidation_listener(self):
        """Listen for database events and invalidate cache"""
        # This would connect to your message broker (Kafka/RabbitMQ)
        # and invalidate cache based on database change events
        pass

Performance Tuning

1. Database Optimization

Connection Pooling:

python

# app/core/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import AsyncAdaptedQueuePool

# Optimized engine configuration
engine = create_async_engine(
    config.DATABASE_URL,
    # Connection pool settings
    poolclass=AsyncAdaptedQueuePool,
    pool_size=20,           # Maximum connections
    max_overflow=30,        # Temporary connections beyond pool_size
    pool_pre_ping=True,     # Verify connections before use
    pool_recycle=3600,      # Recycle connections after 1 hour
    echo=False,             # Set to True for SQL logging in development
    
    # PostgreSQL-specific optimizations
    connect_args={
        "command_timeout": 60,
        "server_settings": {
            "jit": "off",  # JIT can sometimes slow down simple queries
        }
    }
)

# Session factory
AsyncSessionLocal = sessionmaker(
    engine, 
    class_=AsyncSession,
    expire_on_commit=False  # Important for performance
)

Query Optimization:

python

# app/core/query_optimization.py
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload, joinedload

async def get_users_with_orders_optimized(db: AsyncSession):
    """Efficiently load users with their orders"""
    # GOOD: Use selectinload for one-to-many relationships
    stmt = (
        select(User)
        .options(selectinload(User.orders).selectinload(Order.items))
        .limit(100)
    )
    
    result = await db.execute(stmt)
    return result.scalars().all()

async def get_users_paginated(
    db: AsyncSession, 
    page: int = 1, 
    page_size: int = 50
):
    """Efficient pagination using window functions"""
    offset = (page - 1) * page_size
    
    stmt = (
        select(User)
        .order_by(User.created_at.desc())
        .offset(offset)
        .limit(page_size)
    )
    
    result = await db.execute(stmt)
    return result.scalars().all()

# Use indexes effectively
async def create_performance_indexes():
    """Create necessary indexes for performance"""
    indexes = [
        "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_orders_user_id_created_at ON orders(user_id, created_at DESC)",
        "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_users_email ON users(email)",
        "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_order_items_order_id ON order_items(order_id)",
    ]

2. FastAPI Performance Optimizations

Middleware Optimizations:

python

# app/middleware/performance.py
import gzip
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response

class GzipMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        
        # Check if response should be compressed
        if (
            response.status_code == 200 and
            "gzip" in request.headers.get("accept-encoding", "") and
            hasattr(response, "body") and
            len(response.body) > 512  # Only compress larger responses
        ):
            compressed_body = gzip.compress(response.body)
            response.headers["Content-Encoding"] = "gzip"
            response.headers["Content-Length"] = str(len(compressed_body))
            response.body = compressed_body
        
        return response

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        
        response.headers["X-Process-Time"] = str(process_time)
        
        # Log slow requests
        if process_time > 1.0:  # More than 1 second
            logger.warning(
                f"Slow request: {request.method} {request.url.path} "
                f"took {process_time:.2f}s"
            )
        
        return response

Response Model Optimization:

python

# app/models/response_models.py
from pydantic import BaseModel, ConfigDict

class OptimizedResponseModel(BaseModel):
    model_config = ConfigDict(
        from_attributes=True,  # Previously 'orm_mode'
        str_strip_whitespace=True,
        validate_assignment=True,
        # Disable extra fields for performance
        extra='forbid'
    )

class UserResponse(OptimizedResponseModel):
    id: int
    email: str
    name: str
    
    # Use model_validator for complex transformations
    # instead of @property methods which are slower

# Fast response serialization
from fastapi.encoders import jsonable_encoder

async def fast_json_response(data: Any) -> JSONResponse:
    """Optimized JSON response"""
    return JSONResponse(
        content=jsonable_encoder(
            data,
            exclude_none=True,  # Reduce payload size
            custom_encoder={
                # Custom encoders for specific types
                datetime: lambda dt: dt.isoformat(),
            }
        )
    )

3. Advanced Caching Patterns

Write-Through Cache:

python

# app/core/write_through_cache.py
class WriteThroughCache:
    def __init__(self, cache: CacheManager, db: AsyncSession):
        self.cache = cache
        self.db = db
    
    async def create_user(self, user_data: dict) -> User:
        # Write to database first
        user = User(**user_data)
        self.db.add(user)
        await self.db.commit()
        await self.db.refresh(user)
        
        # Then update cache
        cache_key = f"user:{user.id}"
        await self.cache.set(cache_key, user, expire=3600)
        
        # Invalidate related cache entries
        await self.cache.invalidate_pattern("users:*")
        
        return user
    
    async def get_user(self, user_id: int) -> Optional[User]:
        cache_key = f"user:{user_id}"
        
        # Try cache first
        user = await self.cache.get(cache_key)
        if user:
            return user
        
        # Cache miss - get from database
        stmt = select(User).where(User.id == user_id)
        result = await self.db.execute(stmt)
        user = result.scalar_one_or_none()
        
        if user:
            # Update cache for next time
            await self.cache.set(cache_key, user, expire=3600)
        
        return user

Cache Warming:

python

# app/services/cache_warmer.py
class CacheWarmer:
    def __init__(self, cache: CacheManager, db: AsyncSession):
        self.cache = cache
        self.db = db
    
    async def warm_popular_data(self):
        """Pre-load frequently accessed data into cache"""
        tasks = [
            self.warm_popular_products(),
            self.warm_active_users(),
            self.warm_recent_orders(),
        ]
        
        await asyncio.gather(*tasks)
    
    async def warm_popular_products(self):
        """Cache top 1000 popular products"""
        stmt = (
            select(Product)
            .order_by(Product.popularity_score.desc())
            .limit(1000)
        )
        
        result = await self.db.execute(stmt)
        products = result.scalars().all()
        
        for product in products:
            cache_key = f"product:{product.id}"
            await self.cache.set(cache_key, product, expire=7200)
    
    async def schedule_cache_warming(self):
        """Schedule cache warming during low-traffic periods"""
        import asyncio
        
        while True:
            # Run at 3 AM daily
            await asyncio.sleep(self.seconds_until(3, 0))
            await self.warm_popular_data()

4. Monitoring and Metrics

python

# app/core/monitoring.py
import prometheus_client
from prometheus_client import Counter, Histogram, Gauge
from starlette.requests import Request
from starlette.responses import Response

# Metrics
REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint', 'status'])
REQUEST_DURATION = Histogram('http_request_duration_seconds', 'HTTP request duration', ['method', 'endpoint'])
ACTIVE_REQUESTS = Gauge('http_requests_active', 'Active HTTP requests')
CACHE_HITS = Counter('cache_hits_total', 'Cache hits', ['cache_layer'])
CACHE_MISSES = Counter('cache_misses_total', 'Cache misses', ['cache_layer'])

class MetricsMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        ACTIVE_REQUESTS.inc()
        start_time = time.time()
        
        try:
            response = await call_next(request)
        finally:
            ACTIVE_REQUESTS.dec()
        
        duration = time.time() - start_time
        REQUEST_DURATION.labels(
            method=request.method,
            endpoint=request.url.path
        ).observe(duration)
        
        REQUEST_COUNT.labels(
            method=request.method,
            endpoint=request.url.path,
            status=response.status_code
        ).inc()
        
        return response

# Cache metrics decorator
def track_cache_metrics(cache_layer: str):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            try:
                result = await func(*args, **kwargs)
                CACHE_HITS.labels(cache_layer=cache_layer).inc()
                return result
            except CacheMiss:
                CACHE_MISSES.labels(cache_layer=cache_layer).inc()
                raise
        return wrapper
    return decorator

Configuration Summary

Optimal Settings

python

# app/core/performance_config.py
class PerformanceConfig:
    # Database
    DB_POOL_SIZE = 20
    DB_MAX_OVERFLOW = 30
    DB_POOL_RECYCLE = 3600
    
    # Redis
    REDIS_MAX_CONNECTIONS = 50
    REDIS_SOCKET_TIMEOUT = 5
    REDIS_SOCKET_CONNECT_TIMEOUT = 5
    
    # Rate Limiting
    RATE_LIMIT_DEFAULT = (100, 3600)  # 100 requests per hour
    RATE_LIMIT_STRICT = (10, 3600)    # 10 requests per hour for sensitive endpoints
    RATE_LIMIT_GENEROUS = (1000, 3600) # 1000 requests per hour for internal endpoints
    
    # Caching
    CACHE_DEFAULT_TTL = 3600  # 1 hour
    CACHE_SHORT_TTL = 300     # 5 minutes
    CACHE_LONG_TTL = 86400    # 24 hours
    
    # Performance
    MAX_PAGE_SIZE = 100
    DEFAULT_PAGE_SIZE = 50
    ENABLE_GZIP = True
    GZIP_MIN_SIZE = 512

This comprehensive approach to rate limiting, caching, and performance tuning will ensure your FastAPI application can handle high loads efficiently while maintaining reliability and good user experience.