Rate Limiting

Rate Limiting Patterns

Protect APIs with distributed rate limiting using Redis and modern algorithms.

Overview

• Protecting public APIs from abuse
• Implementing tiered rate limits (free/pro/enterprise)
• Scaling rate limiting across multiple instances
• Preventing brute force attacks on auth endpoints
• Managing third-party API consumption

Algorithm Selection

SlowAPI + Redis (FastAPI)

Basic Setup

from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.middleware import SlowAPIMiddleware

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="redis://localhost:6379",
    strategy="moving-window",  # sliding window
)

app = FastAPI()
app.state.limiter = limiter
app.add_middleware(SlowAPIMiddleware)

Endpoint Limits

from slowapi import Limiter

@router.post("/api/v1/auth/login")
@limiter.limit("10/minute")  # Strict for auth
async def login(request: Request, credentials: LoginRequest):
    ...

@router.get("/api/v1/analyses")
@limiter.limit("100/minute")  # Normal for reads
async def list_analyses(request: Request):
    ...

@router.post("/api/v1/analyses")
@limiter.limit("20/minute")  # Moderate for writes
async def create_analysis(request: Request, data: AnalysisCreate):
    ...

User-Based Limits

def get_user_identifier(request: Request) -> str:
    """Rate limit by user ID if authenticated, else IP."""
    if hasattr(request.state, "user"):
        return f"user:{request.state.user.id}"
    return f"ip:{get_remote_address(request)}"

limiter = Limiter(key_func=get_user_identifier)

Token Bucket with Redis (Custom)

import redis.asyncio as redis
from datetime import datetime, timezone

class TokenBucketLimiter:
    def __init__(
        self,
        redis_client: redis.Redis,
        capacity: int = 100,
        refill_rate: float = 10.0,  # tokens per second
    ):
        self.redis = redis_client
        self.capacity = capacity
        self.refill_rate = refill_rate

    async def is_allowed(self, key: str, tokens: int = 1) -> bool:
        """Check if request is allowed, consume tokens atomically."""
        lua_script = """
        local key = KEYS[1]
        local capacity = tonumber(ARGV[1])
        local refill_rate = tonumber(ARGV[2])
        local tokens_requested = tonumber(ARGV[3])
        local now = tonumber(ARGV[4])

        local bucket = redis.call('HMGET', key, 'tokens', 'last_update')
        local current_tokens = tonumber(bucket[1]) or capacity
        local last_update = tonumber(bucket[2]) or now

        -- Calculate refill
        local elapsed = now - last_update
        local refill = elapsed * refill_rate
        current_tokens = math.min(capacity, current_tokens + refill)

        -- Check and consume
        if current_tokens >= tokens_requested then
            current_tokens = current_tokens - tokens_requested
            redis.call('HMSET', key, 'tokens', current_tokens, 'last_update', now)
            redis.call('EXPIRE', key, 3600)
            return 1
        else
            return 0
        end
        """
        now = datetime.now(timezone.utc).timestamp()
        result = await self.redis.eval(
            lua_script, 1, key,
            self.capacity, self.refill_rate, tokens, now
        )
        return result == 1

Sliding Window Counter

class SlidingWindowLimiter:
    def __init__(self, redis_client: redis.Redis, window_seconds: int = 60):
        self.redis = redis_client
        self.window = window_seconds

    async def is_allowed(self, key: str, limit: int) -> tuple[bool, int]:
        """Returns (allowed, remaining)."""
        now = datetime.now(timezone.utc).timestamp()
        window_start = now - self.window

        pipe = self.redis.pipeline()
        # Remove old entries
        pipe.zremrangebyscore(key, 0, window_start)
        # Count current window
        pipe.zcard(key)
        # Add this request
        pipe.zadd(key, {str(now): now})
        # Set expiry
        pipe.expire(key, self.window * 2)

        results = await pipe.execute()
        current_count = results[1]

        if current_count < limit:
            return True, limit - current_count - 1
        return False, 0

Tiered Rate Limits

from enum import Enum

class UserTier(Enum):
    FREE = "free"
    PRO = "pro"
    ENTERPRISE = "enterprise"

TIER_LIMITS = {
    UserTier.FREE: {"requests": 100, "window": 3600},       # 100/hour
    UserTier.PRO: {"requests": 1000, "window": 3600},       # 1000/hour
    UserTier.ENTERPRISE: {"requests": 10000, "window": 3600}, # 10000/hour
}

async def get_rate_limit(user: User) -> str:
    limits = TIER_LIMITS[user.tier]
    return f"{limits['requests']}/{limits['window']}seconds"

@router.get("/api/v1/data")
@limiter.limit(get_rate_limit)
async def get_data(request: Request, user: User = Depends(get_current_user)):
    ...

Response Headers (RFC 6585)

from fastapi import Response

async def add_rate_limit_headers(
    response: Response,
    limit: int,
    remaining: int,
    reset_at: datetime,
):
    response.headers["X-RateLimit-Limit"] = str(limit)
    response.headers["X-RateLimit-Remaining"] = str(remaining)
    response.headers["X-RateLimit-Reset"] = str(int(reset_at.timestamp()))
    response.headers["Retry-After"] = str(int((reset_at - datetime.now(timezone.utc)).seconds))

Error Response (429)

from fastapi import HTTPException
from fastapi.responses import JSONResponse

def rate_limit_exceeded_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=429,
        content={
            "type": "https://api.example.com/errors/rate-limit-exceeded",
            "title": "Too Many Requests",
            "status": 429,
            "detail": "Rate limit exceeded. Please retry after the reset time.",
            "instance": str(request.url),
        },
        headers={
            "Retry-After": "60",
            "X-RateLimit-Limit": "100",
            "X-RateLimit-Remaining": "0",
        }
    )

Anti-Patterns (FORBIDDEN)

# NEVER use in-memory counters in distributed systems
request_counts = {}  # Lost on restart, not shared across instances

# NEVER skip rate limiting on internal APIs (defense in depth)
@router.get("/internal/admin")
async def admin_endpoint():  # No rate limit = vulnerable
    ...

# NEVER use fixed window without considering edge spikes
# A user can hit 100 at 0:59 and 100 at 1:01 = 200 in 2 seconds

Key Decisions

Related Skills

• auth-patterns - Authentication integration
• resilience-patterns - Circuit breakers
• observability-monitoring - Rate limit metrics

Capability Details

token-bucket

Keywords: token bucket, rate limit, burst, capacity Solves:

• How do I implement token bucket rate limiting?
• Allow bursts while limiting rate

sliding-window

Keywords: sliding window, moving window, rate limit Solves:

• How to implement precise rate limiting?
• Avoid fixed window edge cases

slowapi-redis

Keywords: slowapi, fastapi rate limit, redis limiter Solves:

• How to add rate limiting to FastAPI?
• Distributed rate limiting

tiered-limits

Keywords: tiered, user tier, free pro enterprise Solves:

• Different rate limits per subscription tier
• User-based rate limiting

Rate Limiting Implementation Checklist

Planning

• [ ] Define rate limits for each endpoint category
• [ ] Read endpoints (GET) - higher limits
• [ ] Write endpoints (POST/PUT/DELETE) - lower limits
• [ ] Authentication endpoints - very strict limits
• [ ] Expensive operations (LLM calls, file processing) - strictest limits

• [ ] Choose limiting algorithm
• [ ] Token Bucket - for bursty traffic patterns
• [ ] Sliding Window - for strict quotas
• [ ] Fixed Window - for simple requirements

• [ ] Determine key strategy
• [ ] By IP address (anonymous users)
• [ ] By user ID (authenticated users)
• [ ] By API key (service accounts)
• [ ] By organization (enterprise customers)

Implementation

Backend Setup

• [ ] Install dependencies

  pip install slowapi redis

• [ ] Configure Redis connection

  redis_client = Redis.from_url(settings.redis_url)

• [ ] Set up SlowAPI or custom limiter

  limiter = Limiter(key_func=get_user_identifier)
  app.add_middleware(SlowAPIMiddleware)

Route Protection

• [ ] Add @limiter.limit() to all public endpoints
• [ ] Set stricter limits for:
• [ ] Login/register endpoints (prevent brute force)
• [ ] Password reset (prevent enumeration)
• [ ] File upload (prevent abuse)
• [ ] LLM/AI operations (cost control)

Response Headers

• [ ] Include rate limit headers in all responses:
• [ ] X-RateLimit-Limit - max requests in window
• [ ] X-RateLimit-Remaining - requests remaining
• [ ] X-RateLimit-Reset - Unix timestamp when limit resets

• [ ] Include Retry-After header in 429 responses

Error Handling

• [ ] Return proper 429 Too Many Requests status
• [ ] Include helpful error message

  {
    "type": "https://api.example.com/problems/rate-limit-exceeded",
    "title": "Rate Limit Exceeded",
    "status": 429,
    "detail": "You have exceeded 100 requests per minute. Please wait 45 seconds.",
    "retry_after": 45
  }

Tiered Limits

• [ ] Define limits per user tier:

• [ ] Implement dynamic limit function

  def get_tier_limit(request: Request) -> str:
      user = request.state.user
      return TIER_LIMITS.get(user.tier, "10/minute")

Distributed Systems

• [ ] Use Redis backend (not in-memory)
• [ ] Configure Redis connection pooling
• [ ] Set appropriate key TTLs
• [ ] Use Lua scripts for atomicity
• [ ] Handle Redis connection failures gracefully

Monitoring

• [ ] Log rate limit hits

  logger.warning("Rate limit exceeded", extra={
      "user_id": user.id,
      "endpoint": request.url.path,
      "limit": limit,
  })

• [ ] Track metrics:
• [ ] Rate limit hits per endpoint
• [ ] Rate limit hits per user
• [ ] Average remaining quota

• [ ] Set up alerts for:
• [ ] Unusual spike in 429 responses
• [ ] Single user hitting limits repeatedly
• [ ] Redis connection failures

Security Considerations

• [ ] Rate limit login endpoints strictly (prevent brute force)
• [ ] Rate limit password reset (prevent enumeration)
• [ ] Consider IP reputation for anonymous limits
• [ ] Don't expose internal rate limit keys
• [ ] Use secure Redis connection (TLS)

Documentation

• [ ] Document rate limits in OpenAPI/Swagger
• [ ] Add rate limit info to API documentation
• [ ] Include examples of handling 429 responses
• [ ] Explain tier limits for customers

Testing

• [ ] Unit test rate limit logic
• [ ] Integration test with Redis
• [ ] Load test to verify limits work
• [ ] Test retry logic in clients
• [ ] Test header values are correct
• [ ] Test limit reset behavior

Client SDK Recommendations

Document recommended client-side handling:

# Python client example
import time
import httpx

def make_request_with_retry(url: str, max_retries: int = 3):
    for attempt in range(max_retries):
        response = httpx.get(url)

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            time.sleep(retry_after)
            continue

        return response

    raise Exception("Rate limit exceeded after retries")

Rollout Checklist

• [ ] Deploy with monitoring enabled
• [ ] Start with permissive limits
• [ ] Monitor for false positives
• [ ] Gradually tighten limits
• [ ] Communicate changes to users
• [ ] Provide upgrade path for users hitting limits

FastAPI Rate Limiting Examples

Complete examples for implementing rate limiting in FastAPI with Redis.

SlowAPI Setup (Recommended for Simple Cases)

Installation

pip install slowapi redis

Basic Configuration

# app/core/rate_limit.py
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from slowapi.middleware import SlowAPIMiddleware
from redis import Redis

# Use Redis backend for distributed rate limiting
redis_client = Redis.from_url("redis://localhost:6379", decode_responses=True)

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="redis://localhost:6379",
    default_limits=["100/minute"],
)


def setup_rate_limiting(app):
    """Configure rate limiting for the FastAPI app."""
    app.state.limiter = limiter
    app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
    app.add_middleware(SlowAPIMiddleware)

Route-Level Limiting

# app/api/v1/routes/analyses.py
from fastapi import APIRouter, Request, Depends
from slowapi import Limiter
from slowapi.util import get_remote_address

router = APIRouter()
limiter = Limiter(key_func=get_remote_address)

@router.post("/analyses")
@limiter.limit("10/minute")  # Override default
async def create_analysis(request: Request):
    """Create analysis - stricter limit due to resource cost."""
    return {"message": "Analysis created"}

@router.get("/analyses")
@limiter.limit("100/minute")
async def list_analyses(request: Request):
    """List analyses - more permissive."""
    return {"analyses": []}

@router.get("/analyses/{id}")
@limiter.limit("200/minute")
async def get_analysis(request: Request, id: str):
    """Get single analysis - most permissive."""
    return {"id": id}

User-Based Rate Limiting

# app/core/rate_limit.py
from fastapi import Request
from app.api.deps import get_current_user

def get_user_identifier(request: Request) -> str:
    """Get rate limit key from authenticated user or IP."""
    # Try to get user from request state (set by auth middleware)
    user = getattr(request.state, "user", None)
    if user:
        return f"user:{user.id}"

    # Fallback to IP for unauthenticated requests
    return f"ip:{get_remote_address(request)}"

limiter = Limiter(key_func=get_user_identifier)

Tiered Rate Limits

# app/api/v1/routes/protected.py
from fastapi import APIRouter, Request, Depends
from slowapi import Limiter

router = APIRouter()

def get_tier_limit(request: Request) -> str:
    """Dynamic limit based on user tier."""
    user = getattr(request.state, "user", None)
    if not user:
        return "10/minute"  # Anonymous

    tier_limits = {
        "free": "100/minute",
        "pro": "1000/minute",
        "enterprise": "10000/minute",
    }
    return tier_limits.get(user.tier, "100/minute")

@router.post("/generate")
@limiter.limit(get_tier_limit)
async def generate_content(request: Request):
    """Rate limit based on user subscription tier."""
    return {"content": "Generated"}

Custom Redis Token Bucket

For more control, implement custom rate limiting:

# app/core/rate_limit.py
import time
from typing import NamedTuple
import redis.asyncio as redis
from fastapi import Request, HTTPException, status

class RateLimitResult(NamedTuple):
    allowed: bool
    remaining: int
    reset_at: float
    retry_after: int

class RedisRateLimiter:
    """Custom rate limiter with token bucket algorithm."""

    SCRIPT = """
    local key = KEYS[1]
    local capacity = tonumber(ARGV[1])
    local refill_rate = tonumber(ARGV[2])
    local now = tonumber(ARGV[3])

    local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
    local tokens = tonumber(bucket[1]) or capacity
    local last_refill = tonumber(bucket[2]) or now

    local elapsed = (now - last_refill) / 1000
    tokens = math.min(capacity, tokens + elapsed * refill_rate)

    if tokens >= 1 then
        tokens = tokens - 1
        redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
        redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) + 1)
        return {1, math.floor(tokens), 0}
    else
        local retry_after = math.ceil((1 - tokens) / refill_rate)
        redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
        return {0, 0, retry_after}
    end
    """

    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        capacity: int = 100,
        refill_rate: float = 10,
    ):
        self.redis = redis.from_url(redis_url)
        self.capacity = capacity
        self.refill_rate = refill_rate
        self._script = None

    async def _get_script(self):
        if self._script is None:
            self._script = self.redis.register_script(self.SCRIPT)
        return self._script

    async def check(self, key: str) -> RateLimitResult:
        """Check rate limit for a key."""
        script = await self._get_script()
        now_ms = int(time.time() * 1000)

        result = await script(
            keys=[f"ratelimit:{key}"],
            args=[self.capacity, self.refill_rate, now_ms],
        )

        reset_at = time.time() + (self.capacity / self.refill_rate)

        return RateLimitResult(
            allowed=bool(result[0]),
            remaining=int(result[1]),
            reset_at=reset_at,
            retry_after=int(result[2]),
        )


# FastAPI Dependency
async def rate_limit_dependency(
    request: Request,
    limiter: RedisRateLimiter = Depends(get_rate_limiter),
):
    """Dependency that enforces rate limiting."""
    # Get identifier
    user = getattr(request.state, "user", None)
    key = f"user:{user.id}" if user else f"ip:{request.client.host}"

    result = await limiter.check(key)

    # Set rate limit headers
    request.state.rate_limit = result

    if not result.allowed:
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded",
            headers={
                "Retry-After": str(result.retry_after),
                "X-RateLimit-Limit": str(limiter.capacity),
                "X-RateLimit-Remaining": "0",
                "X-RateLimit-Reset": str(int(result.reset_at)),
            },
        )


# Middleware to add rate limit headers to all responses
@app.middleware("http")
async def add_rate_limit_headers(request: Request, call_next):
    response = await call_next(request)

    rate_limit = getattr(request.state, "rate_limit", None)
    if rate_limit:
        response.headers["X-RateLimit-Limit"] = str(100)
        response.headers["X-RateLimit-Remaining"] = str(rate_limit.remaining)
        response.headers["X-RateLimit-Reset"] = str(int(rate_limit.reset_at))

    return response

Usage in Routes

@router.post("/expensive-operation")
async def expensive_operation(
    request: Request,
    _: None = Depends(rate_limit_dependency),
):
    """This endpoint is rate limited."""
    return {"result": "success"}

Rate Limit by Endpoint Cost

# app/core/rate_limit.py
from functools import wraps
from typing import Callable

class CostBasedLimiter:
    """Rate limiter where different operations cost different tokens."""

    def __init__(self, redis_url: str, capacity: int = 1000):
        self.limiter = RedisRateLimiter(redis_url, capacity=capacity)

    def limit(self, cost: int = 1):
        """Decorator that consumes 'cost' tokens per request."""
        def decorator(func: Callable):
            @wraps(func)
            async def wrapper(request: Request, *args, **kwargs):
                key = get_user_identifier(request)

                # Check if we have enough tokens
                for _ in range(cost):
                    result = await self.limiter.check(key)
                    if not result.allowed:
                        raise HTTPException(
                            status_code=429,
                            detail=f"Rate limit exceeded (operation costs {cost} tokens)",
                        )

                return await func(request, *args, **kwargs)
            return wrapper
        return decorator


cost_limiter = CostBasedLimiter("redis://localhost:6379")

@router.get("/simple")
@cost_limiter.limit(cost=1)  # Cheap operation
async def simple_query(request: Request):
    return {"data": "simple"}

@router.post("/generate")
@cost_limiter.limit(cost=10)  # Expensive operation
async def generate_content(request: Request):
    return {"data": "generated"}

@router.post("/bulk-process")
@cost_limiter.limit(cost=50)  # Very expensive
async def bulk_process(request: Request):
    return {"data": "processed"}

Testing Rate Limits

# tests/test_rate_limiting.py
import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_rate_limit_enforced():
    async with AsyncClient(app=app, base_url="http://test") as client:
        # Make requests up to limit
        for _ in range(10):
            response = await client.post("/analyses")
            assert response.status_code == 200

        # Next request should be rate limited
        response = await client.post("/analyses")
        assert response.status_code == 429
        assert "Retry-After" in response.headers

@pytest.mark.asyncio
async def test_rate_limit_headers():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/analyses")

        assert "X-RateLimit-Limit" in response.headers
        assert "X-RateLimit-Remaining" in response.headers
        assert "X-RateLimit-Reset" in response.headers

Related Files

• See references/token-bucket-algorithm.md for algorithm details
• See checklists/rate-limiting-checklist.md for implementation checklist
• See SKILL.md for sliding window and fixed window algorithms

Token Bucket Algorithm

In-depth guide to the token bucket rate limiting algorithm with Redis implementation.

How Token Bucket Works

┌─────────────────────────────────────────────────────────────┐
│                    TOKEN BUCKET                              │
│                                                              │
│   ┌──────────────────────────────────────────────────┐      │
│   │  Tokens: ●●●●●●●○○○  (7/10 tokens available)     │      │
│   │  Capacity: 10 tokens                              │      │
│   │  Refill Rate: 5 tokens/second                     │      │
│   └──────────────────────────────────────────────────┘      │
│                         │                                    │
│   REQUEST ──────────────┼──────────────────────► ALLOWED    │
│                         │                                    │
│   (Each request consumes 1 token)                           │
│   (Bucket refills at constant rate)                         │
│                                                              │
│   Timeline:                                                  │
│   t=0s: 10 tokens │ 10 requests → 0 tokens                  │
│   t=1s: +5 tokens │ 5 tokens available                      │
│   t=2s: +5 tokens │ 10 tokens (capped at capacity)          │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Algorithm Properties

Redis Lua Script (Atomic)

-- token_bucket.lua
-- KEYS[1] = bucket key
-- ARGV[1] = bucket capacity
-- ARGV[2] = refill rate (tokens per second)
-- ARGV[3] = current timestamp (milliseconds)
-- ARGV[4] = tokens to consume

local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])

-- Get current bucket state
local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1]) or capacity
local last_refill = tonumber(bucket[2]) or now

-- Calculate tokens to add based on time elapsed
local elapsed = (now - last_refill) / 1000  -- Convert to seconds
local refill = math.floor(elapsed * refill_rate)
tokens = math.min(capacity, tokens + refill)

-- Check if we can consume tokens
local allowed = 0
local remaining = tokens
local retry_after = 0

if tokens >= requested then
    allowed = 1
    remaining = tokens - requested
else
    -- Calculate when enough tokens will be available
    local needed = requested - tokens
    retry_after = math.ceil(needed / refill_rate)
end

-- Update bucket state
redis.call('HMSET', key,
    'tokens', remaining,
    'last_refill', now
)
redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) + 1)

return {allowed, remaining, retry_after}

Python Implementation

import time
from typing import NamedTuple

import redis.asyncio as redis


class RateLimitResult(NamedTuple):
    allowed: bool
    remaining: int
    retry_after: int  # seconds


class TokenBucket:
    """Token bucket rate limiter with Redis backend."""

    # Load Lua script once
    SCRIPT = """
    local key = KEYS[1]
    local capacity = tonumber(ARGV[1])
    local refill_rate = tonumber(ARGV[2])
    local now = tonumber(ARGV[3])
    local requested = tonumber(ARGV[4])

    local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
    local tokens = tonumber(bucket[1]) or capacity
    local last_refill = tonumber(bucket[2]) or now

    local elapsed = (now - last_refill) / 1000
    local refill = math.floor(elapsed * refill_rate)
    tokens = math.min(capacity, tokens + refill)

    local allowed = 0
    local remaining = tokens
    local retry_after = 0

    if tokens >= requested then
        allowed = 1
        remaining = tokens - requested
    else
        local needed = requested - tokens
        retry_after = math.ceil(needed / refill_rate)
    end

    redis.call('HMSET', key, 'tokens', remaining, 'last_refill', now)
    redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) + 1)

    return {allowed, remaining, retry_after}
    """

    def __init__(
        self,
        redis_client: redis.Redis,
        capacity: int = 100,
        refill_rate: float = 10,  # tokens per second
    ):
        self.redis = redis_client
        self.capacity = capacity
        self.refill_rate = refill_rate
        self._script = self.redis.register_script(self.SCRIPT)

    async def consume(
        self,
        key: str,
        tokens: int = 1,
    ) -> RateLimitResult:
        """
        Try to consume tokens from the bucket.

        Args:
            key: Unique identifier (user_id, ip_address, etc.)
            tokens: Number of tokens to consume

        Returns:
            RateLimitResult with allowed status and metadata
        """
        bucket_key = f"ratelimit:token_bucket:{key}"
        now_ms = int(time.time() * 1000)

        result = await self._script(
            keys=[bucket_key],
            args=[self.capacity, self.refill_rate, now_ms, tokens],
        )

        return RateLimitResult(
            allowed=bool(result[0]),
            remaining=int(result[1]),
            retry_after=int(result[2]),
        )


# Usage with FastAPI
async def get_rate_limiter() -> TokenBucket:
    redis_client = redis.from_url("redis://localhost:6379")
    return TokenBucket(redis_client, capacity=100, refill_rate=10)

Comparison: Token Bucket vs Sliding Window

When to Use Token Bucket

Good for:

• API rate limiting (allows natural bursts)
• User actions (login attempts, form submissions)
• Resource protection (database connections)

Not ideal for:

• Strict per-second quotas
• Billing-based limits (use sliding window)
• Fair queuing (use leaky bucket)

Related Files

• See examples/fastapi-rate-limiting.md for FastAPI integration
• See checklists/rate-limiting-checklist.md for implementation checklist
• See SKILL.md for sliding window implementation