# Rate Limiting

> Rate limiting implementation and strategies

- Skill: `ffsshhttiikk/rate-limiting` (Agent Skill)
- Install (CLI): `npx skillmds@latest add ffsshhttiikk/rate-limiting`
- Raw SKILL.md: https://api.skillmd.com/api/skills/ffsshhttiikk/rate-limiting/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- License: MIT
- Author: ffsshhttiikk (https://skillmd.com/u/ffsshhttiikk)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/ffsshhttiikk/rate-limiting

---

## What I do
- Implement rate limiting
- Choose appropriate algorithms
- Handle rate limit headers
- Design tiered limits
- Handle rate limit exceeded
- Distribute rate limits
- Monitor rate limit usage
- Implement custom providers

## When to use me
When implementing rate limiting or throttling.

## Rate Limiting Algorithms
```python
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, Optional
import asyncio


# Fixed Window
class FixedWindowRateLimiter:
    """
    Simple fixed window rate limiting.
    
    Pros: Simple, memory efficient
    Cons: Can allow bursts at window boundaries
    """
    
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.windows: Dict[str, list] = {}
    
    def is_allowed(self, key: str) -> tuple[bool, dict]:
        now = datetime.utcnow()
        window_start = self._get_window_start(now)
        
        if key not in self.windows:
            self.windows[key] = []
        
        # Remove old requests
        self.windows[key] = [
            t for t in self.windows[key]
            if t >= window_start
        ]
        
        if len(self.windows[key]) >= self.max_requests:
            return False, {
                "limit": self.max_requests,
                "remaining": 0,
                "reset": int((window_start + timedelta(seconds=self.window_seconds)).timestamp())
            }
        
        self.windows[key].append(now)
        
        return True, {
            "limit": self.max_requests,
            "remaining": self.max_requests - len(self.windows[key]),
            "reset": int((window_start + timedelta(seconds=self.window_seconds)).timestamp())
        }
    
    def _get_window_start(self, now: datetime) -> datetime:
        seconds_past_window = now.timestamp() % self.window_seconds
        return now - timedelta(seconds=seconds_past_window)


# Sliding Window
class SlidingWindowRateLimiter:
    """
    Sliding window rate limiting.
    
    Pros: More accurate than fixed window
    Cons: More complex, more memory
    """
    
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests: Dict[str, list] = {}
    
    def is_allowed(self, key: str) -> tuple[bool, dict]:
        now = datetime.utcnow()
        window_start = now - timedelta(seconds=self.window_seconds)
        
        if key not in self.requests:
            self.requests[key] = []
        
        # Remove old requests
        self.requests[key] = [
            t for t in self.requests[key]
            if t >= window_start
        ]
        
        if len(self.requests[key]) >= self.max_requests:
            return False, {
                "limit": self.max_requests,
                "remaining": 0,
                "reset": int((now + timedelta(seconds=self.window_seconds)).timestamp())
            }
        
        self.requests[key].append(now)
        
        return True, {
            "limit": self.max_requests,
            "remaining": self.max_requests - len(self.requests[key]),
            "reset": int((now + timedelta(seconds=self.window_seconds)).timestamp())
        }


# Token Bucket
class TokenBucketRateLimiter:
    """
    Token bucket rate limiting.
    
    Pros: Allows bursts, smooth rate limiting
    Cons: More complex
    """
    
    def __init__(self, max_tokens: int, refill_rate: float):
        self.max_tokens = max_tokens
        self.refill_rate = refill_rate  # tokens per second
        self.tokens: Dict[str, float] = {}
        self.last_refill: Dict[str, datetime] = {}
    
    def _refill(self, key: str) -> float:
        now = datetime.utcnow()
        
        if key not in self.tokens:
            self.tokens[key] = self.max_tokens
            self.last_refill[key] = now
            return self.tokens[key]
        
        elapsed = (now - self.last_refill[key]).total_seconds()
        tokens_to_add = elapsed * self.refill_rate
        
        self.tokens[key] = min(
            self.max_tokens,
            self.tokens[key] + tokens_to_add
        )
        self.last_refill[key] = now
        
        return self.tokens[key]
    
    def is_allowed(self, key: str, tokens: int = 1) -> tuple[bool, dict]:
        self._refill(key)
        
        if self.tokens[key] >= tokens:
            self.tokens[key] -= tokens
            
            return True, {
                "limit": self.max_tokens,
                "remaining": int(self.tokens[key]),
            }
        
        return False, {
            "limit": self.max_tokens": 0,
,
            "remaining        }
```

## API Rate Limiting Middleware
```python
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware


class RateLimitMiddleware(BaseHTTPMiddleware):
    """Rate limiting middleware for FastAPI."""
    
    def __init__(
        self,
        app,
        rate_limiter,
        exempt_paths: list = None
    ):
        super().__init__(app)
        self.rate_limiter = rate_limiter
        self.exempt_paths = exempt_paths or []
    
    async def dispatch(self, request: Request, call_next):
        # Skip rate limiting for exempt paths
        if request.url.path in self.exempt_paths:
            return await call_next(request)
        
        # Get client identifier
        key = self._get_client_key(request)
        
        # Check rate limit
        allowed, headers = self.rate_limiter.is_allowed(key)
        
        if not allowed:
            return JSONResponse(
                status_code=429,
                headers={
                    **headers,
                    "Retry-After": str(headers.get("reset", 60)),
                },
                content={
                    "error": {
                        "code": "RATE_LIMIT_EXCEEDED",
                        "message": "Too many requests. Please retry later."
                    }
                }
            )
        
        response = await call_next(request)
        
        # Add rate limit headers
        for key, value in headers.items():
            response.headers[key] = str(value)
        
        return response
    
    def _get_client_key(self, request: Request) -> str:
        """Get unique key for client."""
        # Check for API key
        api_key = request.headers.get("X-API-Key")
        if api_key:
            return f"apikey:{api_key}"
        
        # Use user ID if authenticated
        if hasattr(request.state, "user_id"):
            return f"user:{request.state.user_id}"
        
        # Fall back to IP
        client_ip = request.client.host
        return f"ip:{client_ip}"
```

## Distributed Rate Limiting with Redis
```python
import redis
from typing import Dict, Any


class RedisRateLimiter:
    """Distributed rate limiter using Redis."""
    
    def __init__(
        self,
        redis_url: str,
        max_requests: int = 100,
        window_seconds: int = 60
    ):
        self.redis = redis.from_url(redis_url)
        self.max_requests = max_requests
        self.window_seconds = window_seconds
    
    def is_allowed(self, key: str) -> tuple[bool, dict]:
        now = datetime.utcnow()
        window_key = f"ratelimit:{key}:{int(now.timestamp() // self.window_seconds)}"
        
        pipe = self.redis.pipeline()
        
        # Increment counter
        pipe.incr(window_key)
        # Set expiry
        pipe.expire(window_key, self.window_seconds * 2)
        # Get current count
        results = pipe.execute()
        
        current_count = results[0]
        
        headers = {
            "X-RateLimit-Limit": str(self.max_requests),
            "X-RateLimit-Remaining": str(max(0, self.max_requests - current_count)),
            "X-RateLimit-Reset": str(
                int(now.timestamp()) + self.window_seconds
            ),
        }
        
        if current_count > self.max_requests:
            return False, headers
        
        return True, headers
    
    def sliding_window(self, key: str) -> tuple[bool, dict]:
        """More accurate sliding window using sorted sets."""
        now = datetime.utcnow()
        window_start = now - timedelta(seconds=self.window_seconds)
        
        pipe = self.redis.pipeline()
        
        # Remove old entries
        pipe.zremrangebyscore(
            f"ratelimit:sliding:{key}",
            0,
            window_start.timestamp() - 1
        )
        
        # Add current request
        pipe.zadd(
            f"ratelimit:sliding:{key}",
            {f"{now.timestamp()}:{uuid.uuid4()}": now.timestamp()}
        )
        
        # Count requests in window
        pipe.zcard(f"ratelimit:sliding:{key}")
        
        # Set expiry
        pipe.expire(
            f"ratelimit:sliding:{key}",
            self.window_seconds * 2
        )
        
        results = pipe.execute()
        current_count = results[2]
        
        allowed = current_count <= self.max_requests
        
        return allowed, {
            "X-RateLimit-Limit": str(self.max_requests),
            "X-RateLimit-Remaining": str(
                max(0, self.max_requests - current_count)
            ),
            "X-RateLimit-Reset": str(
                int(now.timestamp()) + self.window_seconds
            ),
        }
```

## Best Practices
```
Rate Limiting Best Practices:

1. Choose appropriate algorithm
   - Fixed window: simple, forgiving
   - Sliding window: more accurate
   - Token bucket: burst handling

2. Use tiered limits
   - Different limits for different users
   - Public vs authenticated

3. Return proper headers
   - X-RateLimit-Limit
   - X-RateLimit-Remaining
   - X-RateLimit-Reset
   - Retry-After

4. Handle exceeded gracefully
   - Clear error messages
   - Appropriate retry headers

5. Monitor and alert
   - Track rate limit hits
   - Alert on unusual patterns

6. Distribute across instances
   - Use shared storage (Redis)
   - Consistent hashing

7. Consider endpoint-specific
   - Different limits per endpoint
   - Expensive operations get lower limits

8. Allow some burst
   - Users expect occasional bursts
   - Token bucket handles this well

9. Test rate limiting
   - Verify limits are enforced
   - Test header values

10. Document limits
    - API documentation
    - Developer portal
```