Rateon

Python rate-limiting library with Redis backend, async-first design, and framework-agnostic core.

Features

• Async-first - Built with async/await for minimal latency
• Multiple Algorithms - Fixed window, sliding window, token bucket, leaky bucket
• Framework Support - FastAPI and Starlette integrations
• Redis Backend - Atomic operations with Lua scripts, cluster-safe
• In-Memory Fallback - For development and testing
• Observability - Prometheus metrics and structured logging
• Rate Limit Headers - Automatic X-RateLimit-* headers on all responses
• Security - Safe defaults, header spoofing protection, trust proxy support
• Flexible - Per-endpoint, per-router, or global rate limiting

Installation

pip install rateon

Quick Start

Important: Always explicitly configure the backend. Without a config, the library defaults to Redis, which may not be available in development environments.

FastAPI Middleware

from fastapi import FastAPI
from rate_limiter.integrations.fastapi import RateLimiterMiddleware
from rate_limiter.core.rules import Algorithm, RateLimitRule, Scope
from rate_limiter.config import RateLimiterConfig

app = FastAPI()

# Configure backend explicitly
config = RateLimiterConfig(backend="memory", enable_metrics=False)  # Use "redis" in production

app.add_middleware(
    RateLimiterMiddleware,
    rules=[
        RateLimitRule(
            key="ip",
            limit=100,
            window=60,
            algorithm=Algorithm.SLIDING_WINDOW,
            scope=Scope.GLOBAL
        )
    ],
    config=config  # Always specify config
)

@app.get("/")
async def root():
    return {"message": "Hello World"}

FastAPI Decorator

from fastapi import FastAPI
from rate_limiter.integrations.fastapi import rate_limit
from rate_limiter.config import RateLimiterConfig

app = FastAPI()

# IMPORTANT: Always specify config with backend="memory" for development
# Without config, it defaults to Redis which may not be available
config = RateLimiterConfig(backend="memory", enable_metrics=False)

@app.get("/login")
@rate_limit("5/minute", key="ip", config=config)
async def login():
    return {"message": "Login endpoint"}

Important Configuration Notes:

• Always specify config: Without a config, the decorator defaults to Redis backend. If Redis is not available, rate limiting may not work correctly.
• For development/testing: Use RateLimiterConfig(backend="memory", enable_metrics=False)
• For production: Use RateLimiterConfig(backend="redis", redis_url="redis://localhost:6379")

Note: The decorator automatically injects Request for rate limiting, so you don't need to add request: Request to your function signature unless you need to use it in your function.

Custom Time Windows: For custom windows (e.g., 30 minutes, 2 hours), use RateLimitRule objects with middleware instead of the decorator, since rule strings only support standard units (second, minute, hour, day).

FastAPI Dependency

from fastapi import FastAPI, APIRouter, Depends
from rate_limiter.integrations.fastapi import rate_limit_dep
from rate_limiter.config import RateLimiterConfig

app = FastAPI()

# IMPORTANT: Always specify config explicitly!
config = RateLimiterConfig(backend="memory", enable_metrics=False)

router = APIRouter(
    dependencies=[Depends(rate_limit_dep("10/minute", key="ip", config=config))]
)

@router.get("/api/users")
async def get_users():
    return {"users": []}

app.include_router(router)

Rate Limit Headers

Both the middleware and decorator automatically add rate limit headers to all responses, allowing clients to understand their current rate limit status.

Response Headers

The following headers are added to every response:

• X-RateLimit-Limit - The maximum number of requests allowed in the current window
• X-RateLimit-Remaining - The number of requests remaining in the current window
• X-RateLimit-Reset - Unix timestamp (seconds) when the rate limit resets
• Retry-After - Number of seconds until the rate limit resets (only present on 429 responses)

Example Response Headers

Successful Response (200 OK):

X-RateLimit-Limit: 100
X-RateLimit-Remaining: 95
X-RateLimit-Reset: 1704067200

Rate Limited Response (429 Too Many Requests):

Retry-After: 45
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1704067200

Middleware Headers

The middleware automatically adds headers to all responses:

from fastapi import FastAPI
from rate_limiter.integrations.fastapi import RateLimiterMiddleware
from rate_limiter.core.rules import RateLimitRule
from rate_limiter.config import RateLimiterConfig

app = FastAPI()

# Always specify config explicitly
config = RateLimiterConfig(backend="memory", enable_metrics=False)

app.add_middleware(
    RateLimiterMiddleware,
    rules=[RateLimitRule(key="ip", limit=100, window=60)],
    config=config  # Always specify config
)

@app.get("/")
async def root():
    return {"message": "Hello World"}
    # Response will include X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset

Decorator Headers

The decorator also adds headers to responses:

from fastapi import FastAPI, Request
from rate_limiter.integrations.fastapi import rate_limit
from rate_limiter.config import RateLimiterConfig

app = FastAPI()

# Always specify config explicitly
config = RateLimiterConfig(backend="memory", enable_metrics=False)

@app.get("/login")
@rate_limit("5/minute", key="ip", config=config)  # Always pass config
async def login():
    return {"message": "Login endpoint"}
    # Response will include X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset
    # Note: Request is automatically injected by the decorator, so you don't need to add it
    # unless you need to use it in your function

<｜tool▁calls▁begin｜><｜tool▁call▁begin｜> read_file

Note: Headers are automatically added for both successful responses and 429 rate limit errors, providing consistent information to clients about their rate limit status.

Starlette Middleware

from starlette.applications import Starlette
from rate_limiter.integrations.starlette import RateLimiterMiddleware
from rate_limiter.core.rules import Algorithm, RateLimitRule, Scope
from rate_limiter.config import RateLimiterConfig

app = Starlette()

# Configure backend explicitly
config = RateLimiterConfig(backend="memory", enable_metrics=False)  # Use "redis" in production

app.add_middleware(
    RateLimiterMiddleware,
    rules=[
        RateLimitRule(
            key="ip",
            limit=100,
            window=60,
            algorithm=Algorithm.SLIDING_WINDOW,
            scope=Scope.GLOBAL
        )
    ],
    config=config  # Always specify config
)
# All responses will include X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset headers

Global Configuration

from rate_limiter.config import RateLimiterConfig
from rate_limiter.core.limiter import RateLimiter
from rate_limiter.core.rules import RateLimitRule

config = RateLimiterConfig(
    backend="redis",
    redis_url="redis://localhost:6379",
    trust_proxy_headers=True
)

# Create rules using from_string() or explicit parameters
rules = [
    RateLimitRule.from_string("100/minute", key="ip"),  # Use from_string() for rule strings
    # Or: RateLimitRule(key="ip", limit=100, window=60)  # Explicit parameters
]

limiter = RateLimiter(config=config, rules=rules)

Note: When creating RateLimitRule from a string, always use RateLimitRule.from_string(). Do not pass a string directly to the constructor.

Redis Cluster Support

from rate_limiter.config import RateLimiterConfig
from rate_limiter.core.limiter import RateLimiter

# Enable cluster mode
# Only one node URL is needed - client will auto-discover other nodes
config = RateLimiterConfig(
    backend="redis",
    redis_url="redis://node1:6379",  # Single node is sufficient
    redis_cluster_mode=True
)

limiter = RateLimiter(config=config)

For redundancy, you can provide multiple nodes (optional):

config = RateLimiterConfig(
    backend="redis",
    redis_url="redis://node1:6379,redis://node2:6379",  # Optional: multiple nodes for redundancy
    redis_cluster_mode=True
)

Or via environment variable:

RATE_LIMITER_REDIS_CLUSTER_MODE=true
RATE_LIMITER_REDIS_URL=redis://node1:6379

Configuration

Environment Variables

RATE_LIMITER_BACKEND=redis
RATE_LIMITER_REDIS_URL=redis://localhost:6379
RATE_LIMITER_REDIS_CLUSTER_MODE=false
RATE_LIMITER_TRUST_PROXY_HEADERS=true

Python Config

from rate_limiter.config import RateLimiterConfig

config = RateLimiterConfig(
    backend="redis",
    redis_url="redis://localhost:6379",
    default_limits=["100/minute"],
    trust_proxy_headers=True
)

Algorithms

Rateon supports four different rate limiting algorithms, each with different characteristics and use cases.

Fixed Window

How it works: Fixed window divides time into discrete, non-overlapping windows. Each window has a fixed duration (e.g., 60 seconds), and the counter resets at the start of each new window. Requests are counted within the current window, and when the limit is reached, further requests are blocked until the next window begins.

Characteristics:

• Simple implementation with low overhead
• Predictable behavior - limits reset at fixed intervals
• Can allow bursts at window boundaries (e.g., 100 requests at 00:59 and another 100 at 01:00)
• Memory efficient - only needs to track current window count

Use cases:

• Simple rate limiting where occasional bursts are acceptable
• High-throughput scenarios where simplicity matters
• When you need predictable reset times

Example:

from rate_limiter.core.rules import Algorithm, RateLimitRule

RateLimitRule(
    limit=100,
    window=60,
    algorithm=Algorithm.FIXED_WINDOW
)
# Allows up to 100 requests per 60-second window

Sliding Window

How it works: Sliding window tracks requests within a rolling time window. Instead of fixed intervals, it maintains a continuous sliding window that moves forward with each request. Old requests outside the window are removed, and new requests are added. This provides a smooth, continuous rate limit without the boundary burst problem of fixed windows.

Characteristics:

• More accurate than fixed window - no boundary bursts
• Smooth rate limiting that better matches actual request patterns
• Slightly more complex implementation (uses Redis sorted sets)
• Better user experience - more consistent rate limiting

Use cases:

• APIs where smooth rate limiting is important
• Preventing boundary burst attacks
• When you need more accurate rate limiting than fixed window
• User-facing APIs where consistent behavior matters

Example:

from rate_limiter.core.rules import Algorithm, RateLimitRule

RateLimitRule(
    limit=100,
    window=60,
    algorithm=Algorithm.SLIDING_WINDOW
)
# Allows up to 100 requests in any 60-second period

Token Bucket

How it works: Token bucket maintains a bucket of tokens. Tokens are added to the bucket at a constant rate (refill rate). Each request consumes one token. If tokens are available, the request is allowed; otherwise, it's blocked. The bucket has a maximum capacity, allowing bursts up to that capacity while maintaining the average rate over time.

Characteristics:

• Allows controlled bursts up to bucket capacity
• Maintains average rate over time
• Good for handling traffic spikes naturally
• Tokens accumulate when traffic is low, allowing bursts when needed

Use cases:

• APIs that need to handle traffic spikes gracefully
• Services with variable traffic patterns
• When you want to allow bursts but control average rate
• Background job processing with bursty workloads

Example:

from rate_limiter.core.rules import Algorithm, RateLimitRule

RateLimitRule(
    limit=100,  # Bucket capacity (burst size)
    window=60,   # Refill rate: 100 tokens per 60 seconds
    algorithm=Algorithm.TOKEN_BUCKET
)
# Allows bursts up to 100 requests, then refills at ~1.67 requests/second

Leaky Bucket

How it works: Leaky bucket treats requests as water drops entering a bucket. The bucket has a maximum capacity, and it leaks at a constant rate. If the bucket is full, new requests (drops) overflow and are rejected. The bucket continuously leaks at the configured rate, processing requests smoothly at a constant rate regardless of input pattern.

Characteristics:

• Smooth, constant-rate output
• Prevents bursts entirely - enforces strict rate
• Requests are processed at a steady pace
• More restrictive than token bucket - no burst allowance

Use cases:

• APIs that require strict, constant-rate limiting
• Downstream services that can't handle bursts
• When you need to smooth out traffic patterns
• Rate limiting for external API calls

Example:

from rate_limiter.core.rules import Algorithm, RateLimitRule

RateLimitRule(
    limit=100,  # Bucket capacity
    window=60,   # Leak rate: 100 requests per 60 seconds
    algorithm=Algorithm.LEAKY_BUCKET
)
# Processes requests at constant rate of ~1.67 requests/second
# Rejects requests if bucket is full

Algorithm Comparison

Algorithm	Burst Handling	Accuracy	Complexity	Best For
Fixed Window	Allows boundary bursts	Low	Low	Simple use cases
Sliding Window	No bursts	High	Medium	Accurate rate limiting
Token Bucket	Controlled bursts	Medium	Medium	Variable traffic
Leaky Bucket	No bursts	High	Medium	Constant-rate output

Choosing the right algorithm:

• Fixed Window: When simplicity and performance are priorities, and occasional bursts are acceptable
• Sliding Window: When you need accurate, smooth rate limiting without boundary issues
• Token Bucket: When you want to allow bursts but maintain average rate over time
• Leaky Bucket: When you need strict, constant-rate limiting with no burst allowance

Rate Limit Rules

Rate limit rules can be created in two ways:

1. Using Rule String (Simple Format)

For decorators and dependencies, you can use a simple string format:

from rate_limiter.integrations.fastapi import rate_limit, rate_limit_dep

# Format: "limit/unit"
@app.get("/login")
@rate_limit("5/minute", key="ip")  # 5 requests per minute
async def login():
    return {"message": "Login"}

# Available units: second, minute, hour, day
@rate_limit("10/second", key="ip")   # 10 requests per second
@rate_limit("100/hour", key="ip")    # 100 requests per hour
@rate_limit("1000/day", key="ip")    # 1000 requests per day

Rule String Format:

• Format: "limit/unit"
• limit: Positive integer (number of requests)
• unit: One of second, minute, hour, day (case-insensitive)

Examples:

• "5/minute" → 5 requests per 60 seconds
• "10/second" → 10 requests per 1 second
• "100/hour" → 100 requests per 3600 seconds
• "1000/day" → 1000 requests per 86400 seconds

Invalid Examples:

• "5/min" ❌ (must be "minute", not "min")
• "abc/minute" ❌ (limit must be a number)
• "5" ❌ (must include unit: "5/minute")
• "5/minute/hour" ❌ (only one unit allowed)

Limitations of Rule String Format: The rule string format only supports standard units (second, minute, hour, day). For custom time windows like "30 minutes" or "2 hours", you must use RateLimitRule objects directly (see below).

2. Using RateLimitRule Object (Full Control)

For custom time windows (like 30 minutes, 2 hours) or advanced configuration, use the RateLimitRule object:

For middleware and advanced use cases, use the RateLimitRule object:

from rate_limiter.core.rules import Algorithm, RateLimitRule, Scope

rule = RateLimitRule(
    key="ip",                      # Identity key: "ip", "user_id", or custom
    limit=100,                     # Maximum requests
    window=60,                     # Time window in seconds
    algorithm=Algorithm.SLIDING_WINDOW,  # Algorithm to use
    scope=Scope.ENDPOINT           # Scope: ENDPOINT, ROUTER, or GLOBAL
)

Custom Time Windows:

For custom time windows (e.g., 30 minutes, 2 hours), use RateLimitRule with window in seconds:

from rate_limiter.core.rules import RateLimitRule
from rate_limiter.integrations.fastapi import RateLimiterMiddleware

# 10 requests in 30 minutes (1800 seconds)
rule_30min = RateLimitRule(key="ip", limit=10, window=1800)

# 10 requests in 2 hours (7200 seconds)
rule_2hours = RateLimitRule(key="ip", limit=10, window=7200)

# Use with middleware
app.add_middleware(RateLimiterMiddleware, rules=[rule_30min])

Common Time Window Conversions:

• 30 minutes = 1800 seconds (30 * 60)
• 45 minutes = 2700 seconds (45 * 60)
• 2 hours = 7200 seconds (2 * 3600)
• 3 hours = 10800 seconds (3 * 3600)
• 12 hours = 43200 seconds (12 * 3600)

Quick Reference:

• Minutes to seconds: minutes * 60
• Hours to seconds: hours * 3600
• Days to seconds: days * 86400

Examples:

# 10 requests in 30 minutes
RateLimitRule(key="ip", limit=10, window=1800)

# 10 requests in 2 hours
RateLimitRule(key="ip", limit=10, window=7200)

# 5 requests in 45 minutes
RateLimitRule(key="ip", limit=5, window=2700)

Converting String to Rule Object:

You can also convert a rule string to a RateLimitRule object:

from rate_limiter.core.rules import RateLimitRule

# Parse from string - MUST use from_string() method
rule = RateLimitRule.from_string("100/minute", key="ip")
# Equivalent to: RateLimitRule(key="ip", limit=100, window=60)

# Then use with middleware
app.add_middleware(RateLimiterMiddleware, rules=[rule])

Important: You cannot pass a string directly to RateLimitRule() constructor. You must use RateLimitRule.from_string():

# ❌ WRONG - This will raise TypeError
rule = RateLimitRule("5/minute", key="ip")

# ✅ CORRECT - Use from_string() method
rule = RateLimitRule.from_string("5/minute", key="ip")

# ✅ CORRECT - Or use RateLimitRule with explicit parameters
rule = RateLimitRule(key="ip", limit=5, window=60)

Using Custom Windows with Decorators:

For decorators, you can create a RateLimitRule and pass it via a custom function:

from rate_limiter.core.rules import RateLimitRule, Algorithm
from rate_limiter.integrations.fastapi import rate_limit
from rate_limiter.config import RateLimiterConfig

# Create custom rule: 10 requests in 30 minutes
custom_rule = RateLimitRule(key="ip", limit=10, window=1800)

# Create config
config = RateLimiterConfig(backend="memory", enable_metrics=False)

# Use with decorator (requires creating limiter manually)
# Note: Decorators use rule strings, so for custom windows use middleware or dependency
@app.get("/custom")
@rate_limit("10/minute", key="ip", config=config)  # Closest: 10/minute
async def custom():
    return {"message": "Custom endpoint"}

# Better: Use middleware for custom windows
app.add_middleware(
    RateLimiterMiddleware,
    rules=[RateLimitRule(key="ip", limit=10, window=1800)]  # 10 requests in 30 minutes
)

Rule String vs RateLimitRule Object

When to use Rule String:

• Decorators: @rate_limit("5/minute", key="ip")
• Dependencies: rate_limit_dep("10/minute", key="ip")
• Simple use cases where default algorithm (SLIDING_WINDOW) and scope (ENDPOINT) are sufficient

When to use RateLimitRule Object:

• Middleware: Requires RateLimitRule objects
• Custom time windows: Need windows like 30 minutes, 2 hours, 45 minutes, etc. (not supported in rule strings - rule strings only support second, minute, hour, day)
• Custom algorithms: Need to specify algorithm=Algorithm.TOKEN_BUCKET, etc.
• Custom scopes: Need scope=Scope.GLOBAL or Scope.ROUTER
• Multiple rules: Easier to manage with objects
• Priority control: Need to set priority for rule ordering

Example: Using Rule String with Decorator

@app.get("/login")
@rate_limit("5/minute", key="ip")  # Uses default: SLIDING_WINDOW, ENDPOINT scope
async def login():
    return {"message": "Login"}

Example: Using RateLimitRule Object with Middleware

app.add_middleware(
    RateLimiterMiddleware,
    rules=[
        RateLimitRule(
            key="ip",
            limit=100,
            window=60,
            algorithm=Algorithm.TOKEN_BUCKET,  # Custom algorithm
            scope=Scope.GLOBAL,                 # Global scope
            priority=1                          # Custom priority
        )
    ]
)

Available Algorithms

• Algorithm.FIXED_WINDOW - Fixed window algorithm
• Algorithm.SLIDING_WINDOW - Sliding window algorithm (default for rule strings)
• Algorithm.TOKEN_BUCKET - Token bucket algorithm
• Algorithm.LEAKY_BUCKET - Leaky bucket algorithm

Note: When using rule strings (e.g., "5/minute"), the default algorithm is SLIDING_WINDOW. To use a different algorithm, you must use RateLimitRule objects.

Available Scopes

• Scope.ENDPOINT - Apply to individual endpoints (default for rule strings)
• Scope.ROUTER - Apply to all endpoints in a router
• Scope.GLOBAL - Apply globally to all endpoints

Note: When using rule strings (e.g., "5/minute"), the default scope is ENDPOINT. To use a different scope, you must use RateLimitRule objects.

Identity Resolution

Rate limiting can be based on:

• IP Address - key="ip"
• User ID - key="user_id" (requires identity resolver)
• API Key - key="api_key" (requires identity resolver)
• Custom - Provide your own resolver function

Using Built-in Identity Resolvers

The library provides built-in identity resolvers that you can use directly:

IP Address Resolver (default):

from rate_limiter.core.identity import IPIdentityResolver
from rate_limiter.integrations.fastapi import RateLimiterMiddleware

# Default IP resolver
ip_resolver = IPIdentityResolver(trust_proxy=False)

# With proxy support (for behind load balancers)
ip_resolver = IPIdentityResolver(trust_proxy=True)

app.add_middleware(
    RateLimiterMiddleware,
    rules=[...],
    identity_resolver=ip_resolver
)

User ID Resolver:

from rate_limiter.core.identity import UserIdentityResolver
from rate_limiter.integrations.fastapi import rate_limit

# Default user resolver (tries common patterns: request.user, request.state.user, JWT)
user_resolver = UserIdentityResolver()

@app.get("/profile")
@rate_limit("50/hour", key="user_id", identity_resolver=user_resolver)
async def get_profile(request: Request):
    return {"profile": "data"}

API Key Resolver:

from rate_limiter.core.identity import APIKeyIdentityResolver
from rate_limiter.integrations.fastapi import rate_limit

# Default: reads from X-API-Key header
api_key_resolver = APIKeyIdentityResolver()

# Custom header name
api_key_resolver = APIKeyIdentityResolver(header_name="X-Custom-Key")

@app.get("/api/data")
@rate_limit("100/hour", key="api_key", identity_resolver=api_key_resolver)
async def get_data(request: Request):
    return {"data": "protected"}

Note: When using key="ip", key="user_id", or key="api_key" without providing an identity_resolver, the library automatically uses the appropriate built-in resolver. You only need to pass a custom resolver if you want to customize the behavior.

Where to use IdentityResolver:

You can pass an identity_resolver parameter to:

1. Middleware - RateLimiterMiddleware(identity_resolver=...)
2. Decorator - @rate_limit(..., identity_resolver=...)
3. Dependency - rate_limit_dep(..., identity_resolver=...)

Using IdentityResolver with Dependency:

from fastapi import FastAPI, APIRouter, Depends
from rate_limiter.core.identity import UserIdentityResolver
from rate_limiter.integrations.fastapi import rate_limit_dep

app = FastAPI()

# Create a custom resolver
custom_resolver = UserIdentityResolver()

# Use with router dependency
router = APIRouter(
    dependencies=[Depends(rate_limit_dep("10/minute", key="user_id", identity_resolver=custom_resolver))]
)

@router.get("/api/users")
async def get_users():
    return {"users": []}

Custom Identity Resolver

You can create a custom identity resolver by implementing the IdentityResolver protocol. This allows you to extract identity from any source (headers, cookies, request body, etc.).

Example: Custom resolver based on a custom header

from typing import Any
from fastapi import FastAPI, Request
from rate_limiter.integrations.fastapi import RateLimiterMiddleware, rate_limit
from rate_limiter.core.rules import Algorithm, RateLimitRule

app = FastAPI()

class CustomHeaderIdentityResolver:
    """Custom resolver that extracts identity from X-Client-ID header."""
    
    async def resolve(self, request: Any) -> str:
        """Extract client ID from custom header."""
        if hasattr(request, "headers"):
            client_id = request.headers.get("X-Client-ID")
            if client_id:
                return client_id
        return "unknown"

# Use with middleware
custom_resolver = CustomHeaderIdentityResolver()
app.add_middleware(
    RateLimiterMiddleware,
    rules=[RateLimitRule(limit=100, window=60, key="custom", algorithm=Algorithm.FIXED_WINDOW)],
    identity_resolver=custom_resolver
)

# Use with decorator
@app.get("/api/data")
@rate_limit("10/minute", key="custom", identity_resolver=custom_resolver)
async def get_data(request: Request):
    return {"data": "protected"}

Example: Custom resolver using UserIdentityResolver with custom extractor

from fastapi import FastAPI, Request
from rate_limiter.core.identity import UserIdentityResolver
from rate_limiter.integrations.fastapi import rate_limit

app = FastAPI()

def extract_user_id(request):
    """Custom function to extract user ID from request."""
    # Example: Extract from JWT token in Authorization header
    auth_header = request.headers.get("Authorization", "")
    if auth_header.startswith("Bearer "):
        token = auth_header.split(" ")[1]
        # Decode JWT and extract user ID (simplified example)
        # In production, use proper JWT library
        return f"user_{hash(token) % 10000}"
    return "anonymous"

# Create resolver with custom extractor
custom_user_resolver = UserIdentityResolver(user_id_extractor=extract_user_id)

@app.get("/profile")
@rate_limit("50/hour", key="user_id", identity_resolver=custom_user_resolver)
async def get_profile(request: Request):
    return {"profile": "data"}

Example: Composite resolver (multiple identity sources)

from fastapi import FastAPI
from rate_limiter.core.identity import CompositeIdentityResolver, IPIdentityResolver, UserIdentityResolver
from rate_limiter.integrations.fastapi import RateLimiterMiddleware
from rate_limiter.core.rules import Algorithm, RateLimitRule

app = FastAPI()

# Combine IP and User ID for more granular rate limiting
composite_resolver = CompositeIdentityResolver([
    ("ip", IPIdentityResolver(trust_proxy=True)),
    ("user", UserIdentityResolver())
])

# This will create keys like "ip:192.168.1.1:user:12345"
app.add_middleware(
    RateLimiterMiddleware,
    rules=[RateLimitRule(limit=100, window=60, key="composite", algorithm=Algorithm.SLIDING_WINDOW)],
    identity_resolver=composite_resolver
)

Observability

Prometheus Metrics

The library automatically exposes Prometheus metrics:

• rate_limiter_requests_total - Total requests (labels: rule_key, status)
• rate_limiter_requests_allowed - Allowed requests
• rate_limiter_requests_blocked - Blocked requests

Structured Logging

import logging
from rate_limiter.core.limiter import RateLimiter

logger = logging.getLogger("rate_limiter")
# Configure your logging handler

Redis Cluster Support

The library supports both standalone Redis and Redis Cluster:

• Standalone Mode (default): Single Redis instance
• Cluster Mode: Redis Cluster with automatic node discovery
  • Only one node URL is required - the client automatically discovers all other nodes
  • Multiple node URLs can be provided for redundancy (optional)
  • Keys use hash tags to ensure Lua scripts work correctly
  • Automatic failover and slot migration handling

Important:

• In cluster mode, only one node URL is needed. The Redis client will automatically discover the entire cluster topology.
• All keys in Lua scripts must be in the same hash slot. The library automatically uses hash tags ({...}) to ensure this.

Security Considerations

• Safe Redis Keys - Keys are sanitized and prefixed
• Header Spoofing Protection - Only trusted proxies are used for IP resolution
• Fail Closed - On backend failure, requests are denied by default
• Constant-Time Comparison - Prevents timing attacks

Development

# Install development dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Type checking
mypy rate_limiter

# Format code
black rate_limiter tests

# Lint
ruff check rate_limiter tests

License

MIT