# ๐ Fastrict - Enterprise FastAPI Rate Limiter
**The most powerful, flexible, and production-ready rate limiting system for FastAPI applications.**
Fastrict provides enterprise-grade rate limiting with Redis and in-memory backends, supporting everything from simple API throttling to complex multi-tenant rate limiting strategies.
[](https://www.python.org/downloads/)
[](https://fastapi.tiangolo.com/)
[](https://redis.io/)
[](https://opensource.org/licenses/MIT)
[](https://pypi.org/project/fastrict/)
[](https://pypi.org/project/fastrict/)
[](#-performance-benchmarks)
[](#-performance-benchmarks)
## โจ Features
### ๐๏ธ **Dual Architecture Support**
- **๐ Global Rate Limiting**: Shared limits across all endpoints
- **๐ฏ Per-Route Rate Limiting**: Independent limits for each endpoint
- **๐ Hybrid Mode**: Mix global and per-route limits in the same application
### ๐ **Extreme Performance**
- **โก Sub-millisecond latency**: Ultra-fast rate limit checks
- **๐ 1K-30K concurrent connections**: Enterprise-scale performance
- **๐งฎ Sliding window algorithm**: Precise rate limiting with Redis sorted sets
- **๐๏ธ Automatic cleanup**: Expired keys removed automatically
### ๏ฟฝ **Advanced Key Extraction**
- **๐ IP-based limiting**: Traditional client IP throttling
- **๐ Header-based**: API keys, user tokens, custom headers
- **๐ Query parameters**: Rate limit by user ID, tenant, etc.
- **๐ Form fields**: POST form data extraction
- **๐ญ Custom functions**: Complex business logic extraction
- **๐ Combined keys**: Multi-factor rate limiting (IP + API key + tenant)
### ๏ฟฝ๏ธ **Intelligent Bypass System**
- **๐ Role-based bypass**: Skip limits for admin users
- **๐ซ Premium tier bypass**: Different limits for paid users
- **๐ง Maintenance mode**: Conditional bypass during deployments
- **๐ค Custom logic**: Any business rule for bypass decisions
### ๐ **Production Monitoring**
- **๐ Standard HTTP headers**: `X-RateLimit-*` headers
- **๐ฑ Real-time usage**: Current count, remaining, usage percentage
- **โฑ๏ธ Retry-After**: Smart retry timing
- **๐ Comprehensive logging**: Structured logs for monitoring
- **๐ฏ Usage statistics**: Track rate limit effectiveness
### ๐ญ **Enterprise Ready**
- **โ๏ธ Redis Cluster support**: Horizontal scaling
- **๐พ Memory fallback**: In-memory storage for development
- **๐ Graceful degradation**: Continues working if Redis fails
- **๐ Thread-safe**: Concurrent request handling
- **๐งช 100% test coverage**: Thoroughly tested codebase
- **๐ Clean Architecture**: SOLID principles, easy to extend
## ๐ฆ Installation
```bash
# Install from PyPI
pip install fastrict
# Install with development dependencies
pip install fastrict[dev]
# Install with documentation dependencies
pip install fastrict[docs]
```
### ๐ง System Requirements
| Component | Version | Purpose |
|-----------|---------|---------|
| **Python** | 3.8+ | Core runtime |
| **FastAPI** | 0.68+ | Web framework |
| **Redis** | 4.0+ | Primary storage backend |
| **Pydantic** | 1.8+ | Data validation |
| **Starlette** | 0.14+ | ASGI framework |
## ๐ Quick Start
### ๐ฏ 1. Basic Setup (30 seconds)
```python
from fastapi import FastAPI
from fastrict import RateLimitMiddleware, RedisRateLimitRepository
from fastrict import RateLimitUseCase, KeyExtractionUseCase
# Create FastAPI app
app = FastAPI(title="My Rate Limited API")
# Setup rate limiting (Redis)
repository = RedisRateLimitRepository.from_url("redis://localhost:6379")
key_extraction = KeyExtractionUseCase()
rate_limiter = RateLimitUseCase(repository, key_extraction)
# Create default key extraction strategy (NEW in v0.1.0)
# Try API key, then Authorization header, then fall back to IP
from fastrict import create_api_key_fallback
default_key_extraction = create_api_key_fallback()
# Add global rate limiting middleware
app.add_middleware(
RateLimitMiddleware,
rate_limit_use_case=rate_limiter,
excluded_paths=["/health", "/docs", "/metrics"],
default_key_extraction=default_key_extraction # NEW: Default for all routes
)
@app.get("/api/data")
async def get_data():
return {"message": "This endpoint is globally rate limited"}
```
### ๐จ 2. Route-Specific Rate Limiting
```python
from fastrict import throttle, RateLimitStrategyName, RateLimitMode
# Strict rate limiting for authentication
@app.post("/auth/login")
@throttle(strategy=RateLimitStrategyName.SHORT) # 3 requests per minute
async def login():
return {"token": "jwt-token-here"}
# Custom rate limiting for file uploads
@app.post("/api/upload")
@throttle(limit=5, ttl=300) # 5 uploads per 5 minutes
async def upload_file():
return {"file_id": "12345", "status": "uploaded"}
# Premium endpoint with generous limits
@app.get("/api/premium")
@throttle(limit=1000, ttl=3600) # 1000 requests per hour
async def premium_data():
return {"data": "premium content"}
```
### ๐ 3. Advanced Key Extraction
```python
from fastrict import KeyExtractionType
# API key-based rate limiting
@app.get("/api/protected")
@throttle(
limit=100,
ttl=3600,
key_type=KeyExtractionType.HEADER,
key_field="X-API-Key",
key_default="anonymous"
)
async def protected_endpoint():
return {"data": "API key limited content"}
# User-specific rate limiting
@app.get("/api/user-data")
@throttle(
limit=50,
ttl=600,
key_type=KeyExtractionType.QUERY_PARAM,
key_field="user_id",
key_default="guest"
)
async def user_data():
return {"data": "user-specific data"}
# Multi-factor rate limiting (IP + API key)
@app.get("/api/sensitive")
@throttle(
limit=10,
ttl=300,
key_type=KeyExtractionType.COMBINED,
key_combination=["ip", "header:X-API-Key"]
)
async def sensitive_data():
return {"data": "highly sensitive information"}
```
## ๐๏ธ Rate Limiting Modes
Fastrict offers two powerful rate limiting modes that can be mixed and matched:
### ๐ Global Mode
All endpoints share the same rate limit pool. Perfect for overall API protection.
```python
from fastrict import RateLimitMode
app.add_middleware(
RateLimitMiddleware,
rate_limit_use_case=rate_limiter,
rate_limit_mode=RateLimitMode.GLOBAL, # All routes share limits
default_strategy_name=RateLimitStrategyName.MEDIUM
)
@app.get("/api/data") # โโโ
async def get_data(): # โโโ All share same
return {"data": "..."} # โ 20 req/10min pool
@app.get("/api/users") # โ
async def get_users(): # โ
return {"users": []} # โโโ
```
### ๐ฏ Per-Route Mode
Each endpoint has independent rate limit pools. Ideal for fine-grained control.
```python
app.add_middleware(
RateLimitMiddleware,
rate_limit_use_case=rate_limiter,
rate_limit_mode=RateLimitMode.PER_ROUTE # Independent limits per route
)
@app.get("/api/data") # โโ 20 req/10min (independent)
async def get_data():
return {"data": "..."}
@app.get("/api/users") # โโ 20 req/10min (independent)
async def get_users():
return {"users": []}
```
### ๐ Hybrid Mode
Mix global middleware with per-route decorators for ultimate flexibility:
```python
# Global middleware (GLOBAL mode)
app.add_middleware(
RateLimitMiddleware,
rate_limit_use_case=rate_limiter,
rate_limit_mode=RateLimitMode.GLOBAL
)
@app.get("/api/public") # Uses global pool
async def public_data():
return {"data": "public"}
@app.get("/api/special") # Gets its own independent pool
@throttle(limit=100, ttl=3600, rate_limit_mode=RateLimitMode.PER_ROUTE)
async def special_endpoint():
return {"data": "special"}
```
## ๏ฟฝ Fallback Key Extraction Strategies
**NEW in v0.1.0**: Advanced fallback mechanisms that try multiple extraction methods in sequence.
### ๐๏ธ Built-in Fallback Helpers
Fastrict provides convenient helper functions for common fallback patterns:
```python
from fastrict import (
create_auth_header_fallback,
create_api_key_fallback,
create_user_id_fallback
)
# Try Authorization header, then IP
auth_fallback = create_auth_header_fallback(
header_name="Authorization", # Default
default_value="anonymous" # Optional
)
# Try API key, then Authorization, then IP
api_fallback = create_api_key_fallback(
api_key_header="X-API-Key", # Default
auth_header="Authorization", # Default
default_value=None # Will use IP if headers missing
)
# Try user ID from query param, then header, then IP
user_fallback = create_user_id_fallback(
user_id_param="user_id", # Default
user_id_header="X-User-ID", # Default
default_value="anonymous" # Optional
)
```
### โ๏ธ Middleware Default Strategy
Set a default key extraction strategy that applies to all routes:
```python
from fastrict import RateLimitMiddleware, create_api_key_fallback
# Create fallback strategy for middleware
default_strategy = create_api_key_fallback(
api_key_header="X-API-Key",
auth_header="Authorization"
# Falls back to IP if neither header is present
)
app.add_middleware(
RateLimitMiddleware,
rate_limit_use_case=rate_limiter,
default_key_extraction=default_strategy, # Applied to all routes
rate_limit_mode=RateLimitMode.GLOBAL
)
# This endpoint will use the middleware default strategy
@app.get("/api/data")
async def get_data():
return {"data": "Uses API key โ Auth header โ IP fallback"}
# This endpoint overrides with its own strategy
@app.get("/api/users")
@throttle(
limit=50, ttl=3600,
key_extraction_strategy=create_user_id_fallback()
)
async def get_users():
return {"users": "Uses user ID โ header โ IP fallback"}
```
### ๐ฏ Route-Specific Fallback
Override the middleware default for specific routes:
```python
# Use helper function directly
@app.get("/api/auth-required")
@throttle(
limit=100, ttl=3600,
key_extraction_strategy=create_auth_header_fallback()
)
async def auth_endpoint():
return {"data": "auth-protected"}
# Custom fallback strategy
from fastrict import KeyExtractionStrategy, KeyExtractionType
custom_fallback = KeyExtractionStrategy(
type=KeyExtractionType.FALLBACK,
fallback_strategies=[
KeyExtractionStrategy(
type=KeyExtractionType.HEADER,
field_name="X-Session-ID"
),
KeyExtractionStrategy(
type=KeyExtractionType.HEADER,
field_name="X-API-Key"
),
KeyExtractionStrategy(
type=KeyExtractionType.IP
)
]
)
@app.get("/api/session-data")
@throttle(
limit=50, ttl=600,
key_extraction_strategy=custom_fallback
)
async def session_endpoint():
return {"data": "session-based rate limiting"}
```
### ๐ How Fallback Works
1. **Try first strategy**: Attempt to extract key using the first method
2. **Check success**: If extraction succeeds and returns a valid key, use it
3. **Try next strategy**: If extraction fails or returns empty, try next method
4. **Continue sequence**: Repeat until a strategy succeeds
5. **IP fallback**: If all strategies fail, fall back to IP address
```python
# Example: API key โ Auth header โ IP fallback
api_fallback = create_api_key_fallback()
# For a request with these headers:
# X-API-Key: "" (empty)
# Authorization: "Bearer token123"
# Client IP: "192.168.1.100"
# Fallback process:
# 1. Try X-API-Key โ empty, skip
# 2. Try Authorization โ "Bearer token123" โ
# Result: Rate limiting key = "Bearer token123"
```
### ๐ข Real-World Example
```python
# Multi-tenant SaaS with intelligent key extraction
from fastrict import create_api_key_fallback, RateLimitMode
# Middleware default: API key for tenant isolation
default_strategy = create_api_key_fallback(
api_key_header="X-API-Key",
auth_header="Authorization"
)
app.add_middleware(
RateLimitMiddleware,
rate_limit_use_case=rate_limiter,
default_key_extraction=default_strategy,
rate_limit_mode=RateLimitMode.GLOBAL,
default_strategy_name=RateLimitStrategyName.MEDIUM
)
# Public endpoints use IP-based limiting
@app.get("/api/public")
@throttle(
limit=100, ttl=3600,
key_extraction_strategy=KeyExtractionStrategy(type=KeyExtractionType.IP)
)
async def public_data():
return {"data": "public"}
# User endpoints prefer user ID over API key
@app.get("/api/user-profile")
@throttle(
limit=200, ttl=3600,
key_extraction_strategy=create_user_id_fallback()
)
async def user_profile():
return {"profile": "user data"}
# Admin endpoints use session-based limiting
admin_fallback = KeyExtractionStrategy(
type=KeyExtractionType.FALLBACK,
fallback_strategies=[
KeyExtractionStrategy(type=KeyExtractionType.HEADER, field_name="Admin-Session"),
KeyExtractionStrategy(type=KeyExtractionType.HEADER, field_name="X-API-Key"),
KeyExtractionStrategy(type=KeyExtractionType.IP)
]
)
@app.get("/api/admin")
@throttle(
limit=1000, ttl=3600,
key_extraction_strategy=admin_fallback
)
async def admin_endpoint():
return {"data": "admin-only"}
```
## ๏ฟฝ๐ Key Extraction Strategies
### ๐ IP-Based (Default)
```python
@throttle(limit=100, ttl=3600) # Rate limit per client IP
```
### ๐ซ Header-Based
```python
# API key rate limiting
@throttle(
limit=1000, ttl=3600,
key_type=KeyExtractionType.HEADER,
key_field="X-API-Key",
key_default="anonymous"
)
# User token rate limiting
@throttle(
limit=500, ttl=3600,
key_type=KeyExtractionType.HEADER,
key_field="Authorization",
key_default="unauthenticated"
)
```
### ๐ Query Parameter-Based
```python
# User-specific limits
@throttle(
limit=200, ttl=3600,
key_type=KeyExtractionType.QUERY_PARAM,
key_field="user_id",
key_default="anonymous"
)
# Tenant-based limits (SaaS)
@throttle(
limit=10000, ttl=3600,
key_type=KeyExtractionType.QUERY_PARAM,
key_field="tenant_id",
key_default="free_tier"
)
```
### ๐ Combined Key Strategies
```python
# Multi-factor rate limiting
@throttle(
limit=50, ttl=300,
key_type=KeyExtractionType.COMBINED,
key_combination=[
"ip", # Client IP
"header:X-API-Key", # API key
"query_param:tenant_id" # Tenant
]
)
# Results in key: "192.168.1.1:abc123:tenant_456"
```
### ๐ญ Custom Key Extraction
```python
def extract_session_key(request: Request) -> str:
"""Complex business logic for key extraction."""
session_id = request.headers.get("Session-ID")
user_tier = request.headers.get("User-Tier", "free")
if user_tier == "premium":
return f"premium:session:{session_id}"
elif user_tier == "enterprise":
return f"enterprise:session:{session_id}"
else:
return f"free:ip:{request.client.host}"
@throttle(
limit=100, ttl=3600,
key_type=KeyExtractionType.CUSTOM,
key_extractor=extract_session_key
)
async def complex_endpoint():
return {"data": "complex rate limiting"}
```
## ๐ก๏ธ Smart Bypass System
Create intelligent bypass rules for different user roles, maintenance modes, or business logic.
### ๐ Role-Based Bypass
```python
def bypass_for_admins(request: Request) -> bool:
"""Bypass rate limiting for admin users."""
user_role = request.headers.get("User-Role")
return user_role in ["admin", "superuser"]
@app.get("/api/admin-only")
@throttle(
limit=10, ttl=60,
bypass_function=bypass_for_admins,
custom_error_message="Admin endpoint requires admin privileges"
)
async def admin_endpoint():
return {"data": "admin-only data"}
```
### ๐ซ Premium User Bypass
```python
def bypass_for_premium(request: Request) -> bool:
"""Bypass limits for premium subscribers."""
subscription = request.headers.get("Subscription-Tier")
return subscription in ["premium", "enterprise"]
@app.get("/api/premium-features")
@throttle(
limit=5, ttl=60, # Limits for free users
bypass_function=bypass_for_premium
)
async def premium_features():
return {"features": ["advanced", "priority"]}
```
### ๐ง Maintenance Mode Bypass
```python
import os
def bypass_during_maintenance(request: Request) -> bool:
"""Bypass rate limiting during maintenance."""
maintenance_mode = os.getenv("MAINTENANCE_MODE", "false").lower() == "true"
maintenance_key = request.headers.get("Maintenance-Key")
return maintenance_mode and maintenance_key == os.getenv("MAINTENANCE_SECRET")
@app.get("/api/critical")
@throttle(
limit=100, ttl=3600,
bypass_function=bypass_during_maintenance
)
async def critical_endpoint():
return {"data": "critical system data"}
```
## ๐ Built-in Strategies
Fastrict comes with pre-configured strategies for common use cases:
```python
from fastrict import RateLimitStrategy, RateLimitStrategyName
# Define custom strategies
custom_strategies = [
RateLimitStrategy(
name=RateLimitStrategyName.SHORT,
limit=3,
ttl=60
), # Strict: 3 requests per minute
RateLimitStrategy(
name=RateLimitStrategyName.MEDIUM,
limit=20,
ttl=600
), # Moderate: 20 requests per 10 minutes
RateLimitStrategy(
name=RateLimitStrategyName.LONG,
limit=100,
ttl=3600
), # Generous: 100 requests per hour
]
app.add_middleware(
RateLimitMiddleware,
rate_limit_use_case=rate_limiter,
default_strategies=custom_strategies,
default_strategy_name=RateLimitStrategyName.MEDIUM
)
# Use predefined strategies
@app.post("/auth/login")
@throttle(strategy=RateLimitStrategyName.SHORT) # Use strict limits
async def login():
return {"message": "Login attempt"}
@app.get("/api/search")
@throttle(strategy=RateLimitStrategyName.LONG) # Use generous limits
async def search():
return {"results": []}
```
## ๐๏ธ Storage Backends
### โก Redis Backend (Recommended)
Perfect for production, supports clustering and persistence.
```python
from fastrict import RedisRateLimitRepository
# Simple connection
repository = RedisRateLimitRepository.from_url("redis://localhost:6379")
# Advanced configuration
repository = RedisRateLimitRepository.from_url(
redis_url="redis://:password@localhost:6379/0",
key_prefix="myapp_limits",
logger=my_logger
)
# Custom Redis client
import redis
redis_client = redis.Redis(
host="localhost",
port=6379,
password="secret",
decode_responses=True,
socket_timeout=5,
retry_on_timeout=True
)
repository = RedisRateLimitRepository(
redis_client=redis_client,
key_prefix="production_limits"
)
```
### ๐พ Memory Backend (Development)
Great for testing and development environments.
```python
from fastrict import MemoryRateLimitRepository
# In-memory storage (no persistence)
repository = MemoryRateLimitRepository(
key_prefix="dev_limits",
cleanup_interval=300 # Cleanup every 5 minutes
)
```
## ๐ Monitoring & Observability
### ๐ Standard HTTP Headers
Fastrict automatically adds industry-standard rate limiting headers:
```http
HTTP/1.1 200 OK
X-RateLimit-Limit: 100 # Maximum requests in window
X-RateLimit-Remaining: 75 # Requests remaining in window
X-RateLimit-Used: 25 # Requests used in window
X-RateLimit-Window: 3600 # Window duration in seconds
```
When rate limited (HTTP 429):
```http
HTTP/1.1 429 Too Many Requests
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Used: 100
X-RateLimit-Window: 3600
Retry-After: 1847 # Seconds until window resets
```
### ๐ฑ Real-time Status Endpoint
```python
@app.get("/api/rate-limit-status")
@throttle(bypass=True) # Don't count status checks against limits
async def rate_limit_status(request: Request):
"""Get current rate limit status without incrementing counter."""
result = rate_limiter.get_current_usage(
request=request,
middleware_rate_limit_mode=RateLimitMode.GLOBAL,
route_path=request.url.path
)
return {
"allowed": result.allowed,
"current_count": result.current_count,
"limit": result.limit,
"remaining": result.remaining_requests,
"reset_in_seconds": result.ttl,
"usage_percentage": result.usage_percentage,
"strategy": result.strategy_name,
"key": result.key # Rate limiting key used
}
```
### ๐ Structured Error Responses
```json
{
"message": "Rate limit exceeded. Maximum 100 requests per 3600 seconds. Please try again in 1847 seconds.",
"retry_after": 1847,
"limit": 100,
"window": 3600,
"current_count": 100,
"usage_percentage": 100.0,
"strategy": "medium"
}
```
### ๐ง Custom Error Messages
```python
@app.post("/api/critical")
@throttle(
limit=5, ttl=60,
custom_error_message="Critical endpoint allows only 5 requests per minute. Please use batch operations for bulk requests."
)
async def critical_operation():
return {"status": "processing"}
```
## ๐งช Testing Your Rate Limits
### ๐ Unit Testing
```python
import pytest
from fastapi.testclient import TestClient
from unittest.mock import Mock
def test_rate_limiting():
# Mock Redis for testing
mock_redis = Mock()
repository = RedisRateLimitRepository(mock_redis)
with TestClient(app) as client:
# First request should succeed
response = client.get("/api/data")
assert response.status_code == 200
assert "X-RateLimit-Remaining" in response.headers
# Simulate rate limit exceeded
mock_redis.zcard.return_value = 100 # Over limit
response = client.get("/api/data")
assert response.status_code == 429
assert "Retry-After" in response.headers
```
### ๐ Integration Testing
```python
import asyncio
import httpx
async def test_concurrent_requests():
"""Test rate limiting under concurrent load."""
async with httpx.AsyncClient() as client:
# Fire 10 concurrent requests
tasks = [
client.get("http://localhost:8000/api/data")
for _ in range(10)
]
responses = await asyncio.gather(*tasks)
# Check that some are rate limited
success_count = sum(1 for r in responses if r.status_code == 200)
rate_limited_count = sum(1 for r in responses if r.status_code == 429)
assert success_count <= 5 # Our test limit
assert rate_limited_count >= 5
```
### ๐จ Load Testing
```bash
# Install hey for load testing
go install github.com/rakyll/hey@latest
# Test rate limiting under load
hey -n 100 -c 10 -H "X-API-Key: test123" http://localhost:8000/api/data
# Expected output shows rate limiting in action:
# Status code distribution:
# [200] 20 responses (successful requests)
# [429] 80 responses (rate limited)
```
## ๐ Performance Characteristics
### โก Benchmarks
| Metric | Value | Notes |
|--------|-------|-------|
| **Latency** | < 1ms | Rate limit check overhead |
| **Throughput** | 30K+ req/s | Redis backend, single instance |
| **Memory** | ~10MB | Per 100K active keys |
| **CPU** | < 1% | Minimal overhead |
### ๐ Scalability
```python
# Horizontal scaling with Redis Cluster
repository = RedisRateLimitRepository.from_url(
"redis://node1:7000,node2:7000,node3:7000",
key_prefix="cluster_limits"
)
# Multiple app instances can share rate limits
# Perfect for microservices and load-balanced deployments
```
## ๐๏ธ Architecture & Design
Fastrict follows **Clean Architecture** principles:
```
src/fastrict/
โโโ entities/ # ๐๏ธ Core business models & enums
โ โโโ models.py # RateLimitStrategy, RateLimitResult
โ โโโ enums.py # KeyExtractionType, RateLimitMode
โโโ use_cases/ # ๐ง Business logic & orchestration
โ โโโ rate_limit.py # Core rate limiting logic
โ โโโ key_extraction.py Key extraction strategies
โโโ adapters/ # ๐ External integrations
โ โโโ redis_repository.py Redis storage backend
โ โโโ memory_repository.py In-memory storage backend
โโโ frameworks/ # ๐ FastAPI integration
โโโ middleware.py # Global rate limiting middleware
โโโ decorator.py # @throttle route decorator
```
### ๐ฏ Design Principles
- **๐ Immutable Entities**: Thread-safe by design
- **๐งช Dependency Injection**: Easy testing and mocking
- **๐ Interface Segregation**: Swap backends seamlessly
- **๐ฆ Single Responsibility**: Each component has one job
- **๐ Performance First**: Optimized for high throughput
## ๐ Performance Benchmarks
*Last updated: 2025-10-02 (MacOS 26, M3 Pro, conda chat environment)*
Fastrict has been extensively tested for performance under various load conditions. Here are the benchmark results:
### โก Single Request Performance
| Metric | Value | Description |
|--------|-------|-------------|
| **Single Request Latency** | **0.37 ms** | Ultra-fast rate limit check overhead |
### ๐โโ๏ธ Sequential Performance
| Metric | Value | Description |
|--------|-------|-------------|
| **Total Requests** | 1,000 | Sequential test requests |
| **Duration** | 0.35 seconds | Total test time |
| **Requests/Second** | **2,857 RPS** | Sequential throughput |
| **Average Response Time** | **0.35 ms** | Mean response time |
| **P95 Response Time** | **0.41 ms** | 95th percentile |
### ๐ Concurrent Performance (High Load)
| Metric | Value | Description |
|--------|-------|-------------|
| **Total Requests** | 1,000 | 50 users ร 20 requests each |
| **Duration** | 0.27 seconds | Concurrent execution time |
| **Requests/Second** | **3,676 RPS** | Concurrent throughput |
| **Success Rate** | **100.0%** | Zero failures under load |
| **Average Response Time** | **13.41 ms** | Mean response time |
| **P95 Response Time** | **28.64 ms** | 95th percentile |
| **P99 Response Time** | **28.93 ms** | 99th percentile |
### ๐ก๏ธ Rate Limiting Accuracy
| Metric | Value | Description |
|--------|-------|-------------|
| **Total Requests** | 100 | Concurrent requests to limited endpoint |
| **Successful Requests** | 50 | Requests within limit |
| **Rate Limited Requests** | 50 | Correctly blocked requests |
| **Accuracy** | **100%** | Perfect rate limiting enforcement |
| **Average Response Time** | **10.37 ms** | Fast even when blocking |
### ๐ช Extreme Load Test
| Metric | Value | Description |
|--------|-------|-------------|
| **Total Requests** | 1,000 | 100 users ร 10 requests each |
| **Requests/Second** | **3,639 RPS** | Sustained under extreme load |
| **Success Rate** | **100.0%** | No failures under pressure |
| **Error Rate** | **0.0%** | System stability maintained |
| **P99 Response Time** | **32.56 ms** | Excellent tail latency |
### ๐ Sustained Load Endurance
| Metric | Value | Description |
|--------|-------|-------------|
| **Total Requests** | 913 | 10-second endurance test |
| **Achieved RPS** | **91.24** | Target: 100 RPS |
| **Success Rate** | **100.0%** | No degradation over time |
| **Average Response Time** | **1.95 ms** | Consistent performance |
| **Performance Degradation** | **21.7%** | Minimal performance loss |
### ๐ Performance Highlights
- โก **Sub-millisecond latency**: 0.37ms average response time
- ๐ **3,600+ RPS**: Exceptional concurrent throughput
- ๐ฏ **100% success rate**: Perfect stability under load
- ๐ก๏ธ **100% rate limiting accuracy**: Precise enforcement
- ๐พ **Memory efficient**: Handles thousands of unique keys
- ๐ **Minimal degradation**: Stable performance over time
### ๐งช Test Environment
- **Hardware**: MacOS, M1 Pro
- **Python**: 3.10.16 (conda environment)
- **Backend**: In-memory storage (optimal performance)
- **Test Framework**: pytest + httpx + asyncio
- **Load Patterns**: Sequential, concurrent, sustained, extreme scenarios
### ๐ฌ Run Performance Tests Yourself
Want to verify these results? Run the performance tests on your own system:
```bash
# Install dependencies
conda activate chat # or your preferred environment
pip install pytest httpx pytest-asyncio uvicorn
pip install -e .
# Run comprehensive performance test suite
python -m pytest tests/test_performance.py -v
# Run live performance demo
python test/demo_performance.py
# Generate performance report
python test/run_performance_tests.py
```
See [`PERFORMANCE_SUMMARY.md`](PERFORMANCE_SUMMARY.md) and [`tests/PERFORMANCE.md`](tests/PERFORMANCE.md) for detailed testing documentation.
### ๐ Real-World Performance
These benchmarks demonstrate that Fastrict can easily handle:
- **High-traffic APIs**: 3,000+ requests per second
- **Real-time applications**: Sub-millisecond response times
- **Microservices**: Zero performance impact
- **Enterprise workloads**: 100% stability under pressure
*Performance may vary based on hardware, Redis configuration, and network conditions.*
## ๐ฏ Real-World Examples
### ๐ข Multi-Tenant SaaS Application
```python
def extract_tenant_key(request: Request) -> str:
"""Extract tenant-aware rate limiting key."""
api_key = request.headers.get("X-API-Key", "")
tenant_id = request.headers.get("X-Tenant-ID", "unknown")
# Different limits based on subscription tier
if api_key.startswith("ent_"):
return f"enterprise:tenant:{tenant_id}"
elif api_key.startswith("pro_"):
return f"professional:tenant:{tenant_id}"
else:
return f"free:tenant:{tenant_id}"
# Different strategies per tier
enterprise_strategy = RateLimitStrategy(name=RateLimitStrategyName.CUSTOM, limit=10000, ttl=3600)
professional_strategy = RateLimitStrategy(name=RateLimitStrategyName.LONG, limit=1000, ttl=3600)
free_strategy = RateLimitStrategy(name=RateLimitStrategyName.MEDIUM, limit=100, ttl=3600)
@app.get("/api/analytics")
@throttle(
limit=100, # Free tier limit
ttl=3600,
key_type=KeyExtractionType.CUSTOM,
key_extractor=extract_tenant_key
)
async def get_analytics():
return {"analytics": "tenant-specific data"}
```
### ๐ E-commerce API Protection
```python
# Protect checkout process
@app.post("/api/checkout")
@throttle(
limit=5, ttl=300, # 5 checkouts per 5 minutes
key_type=KeyExtractionType.HEADER,
key_field="User-ID",
custom_error_message="Too many checkout attempts. Please wait before trying again."
)
async def process_checkout():
return {"order_id": "12345", "status": "processing"}
# Protect payment endpoints with combined key (user + IP)
@app.post("/api/payment")
@throttle(
limit=3, ttl=600, # 3 payment attempts per 10 minutes
key_type=KeyExtractionType.COMBINED,
key_combination=["header:User-ID", "ip"],
custom_error_message="Payment rate limit exceeded. Contact support if you need assistance."
)
async def process_payment():
return {"payment_id": "pay_123", "status": "success"}
```
### ๐ Authentication & Security
```python
# Login rate limiting with exponential backoff
@app.post("/auth/login")
@throttle(
limit=5, ttl=900, # 5 login attempts per 15 minutes
key_type=KeyExtractionType.COMBINED,
key_combination=["ip", "form_field:username"],
custom_error_message="Too many login attempts. Account temporarily locked."
)
async def login():
return {"token": "jwt_token", "expires_in": 3600}
# Password reset protection
@app.post("/auth/password-reset")
@throttle(
limit=3, ttl=3600, # 3 password resets per hour
key_type=KeyExtractionType.FORM_FIELD,
key_field="email",
custom_error_message="Password reset limit exceeded. Try again in an hour."
)
async def password_reset():
return {"message": "Password reset email sent"}
# 2FA verification
@app.post("/auth/verify-2fa")
@throttle(
limit=10, ttl=300, # 10 attempts per 5 minutes
key_type=KeyExtractionType.HEADER,
key_field="Session-ID",
custom_error_message="Too many 2FA verification attempts."
)
async def verify_2fa():
return {"verified": True}
```
### ๐ฑ Mobile API with Device Limits
```python
def extract_device_key(request: Request) -> str:
"""Rate limit by device fingerprint."""
device_id = request.headers.get("Device-ID")
app_version = request.headers.get("App-Version", "unknown")
platform = request.headers.get("Platform", "unknown")
if device_id:
return f"device:{device_id}:{platform}:{app_version}"
else:
return f"ip:{request.client.host}"
@app.get("/api/mobile/sync")
@throttle(
limit=100, ttl=3600, # 100 syncs per hour per device
key_type=KeyExtractionType.CUSTOM,
key_extractor=extract_device_key
)
async def mobile_sync():
return {"sync_data": "device-specific data"}
```
### ๐ค Bot Protection & Scraping Prevention
```python
def detect_bot(request: Request) -> bool:
"""Detect and allow verified bots."""
user_agent = request.headers.get("User-Agent", "").lower()
bot_token = request.headers.get("Bot-Token")
# Allow verified search engine bots
verified_bots = ["googlebot", "bingbot", "slurp"]
if any(bot in user_agent for bot in verified_bots):
return True
# Allow bots with valid tokens
return bot_token in os.getenv("VALID_BOT_TOKENS", "").split(",")
@app.get("/api/public-data")
@throttle(
limit=10, ttl=60, # Strict limits for non-bots
bypass_function=detect_bot,
key_type=KeyExtractionType.COMBINED,
key_combination=["ip", "header:User-Agent"]
)
async def public_data():
return {"data": "public information"}
```
## ๐ง Configuration Examples
### ๐ Environment-Based Configuration
```python
import os
from fastrict import RateLimitStrategy, RateLimitStrategyName
def get_rate_limit_config():
"""Get rate limit configuration based on environment."""
env = os.getenv("ENVIRONMENT", "development")
if env == "production":
return {
"strategies": [
RateLimitStrategy(name=RateLimitStrategyName.SHORT, limit=5, ttl=60),
RateLimitStrategy(name=RateLimitStrategyName.MEDIUM, limit=50, ttl=600),
RateLimitStrategy(name=RateLimitStrategyName.LONG, limit=500, ttl=3600),
],
"redis_url": os.getenv("REDIS_URL"),
"key_prefix": "prod_limits"
}
elif env == "staging":
return {
"strategies": [
RateLimitStrategy(name=RateLimitStrategyName.SHORT, limit=10, ttl=60),
RateLimitStrategy(name=RateLimitStrategyName.MEDIUM, limit=100, ttl=600),
RateLimitStrategy(name=RateLimitStrategyName.LONG, limit=1000, ttl=3600),
],
"redis_url": os.getenv("REDIS_URL", "redis://localhost:6379/1"),
"key_prefix": "staging_limits"
}
else: # development
return {
"strategies": [
RateLimitStrategy(name=RateLimitStrategyName.SHORT, limit=100, ttl=60),
RateLimitStrategy(name=RateLimitStrategyName.MEDIUM, limit=1000, ttl=600),
RateLimitStrategy(name=RateLimitStrategyName.LONG, limit=10000, ttl=3600),
],
"redis_url": "redis://localhost:6379/0",
"key_prefix": "dev_limits"
}
# Apply configuration
config = get_rate_limit_config()
repository = RedisRateLimitRepository.from_url(
redis_url=config["redis_url"],
key_prefix=config["key_prefix"]
)
app.add_middleware(
RateLimitMiddleware,
rate_limit_use_case=rate_limiter,
default_strategies=config["strategies"],
default_strategy_name=RateLimitStrategyName.MEDIUM
)
```
### ๐ Feature Flags Integration
```python
def feature_flag_bypass(request: Request) -> bool:
"""Bypass rate limiting based on feature flags."""
# Integration with feature flag service
user_id = request.headers.get("User-ID")
if user_id:
# Check if user has rate limiting bypass feature enabled
return feature_flag_service.is_enabled(
flag="rate_limiting_bypass",
user_id=user_id
)
return False
@app.get("/api/experimental")
@throttle(
limit=10, ttl=300,
bypass_function=feature_flag_bypass
)
async def experimental_feature():
return {"feature": "experimental"}
```
## ๐ค Contributing
We welcome contributions! Fastrict is built with โค๏ธ by the community.
### ๐ Quick Start for Contributors
```bash
# Fork and clone the repository
git clone https://github.com/yourusername/fastrict.git
cd fastrict
# Install development dependencies
pip install -e ".[dev]"
# Run tests
pytest
# Run linting
black src tests
flake8 src tests
mypy src
# Run the example
python src/examples/simple_example.py
```
### ๐ Contribution Guidelines
- **๐ Bug Reports**: Use the issue tracker with detailed reproduction steps
- **โจ Feature Requests**: Propose new features with use cases
- **๐ Documentation**: Help improve our docs and examples
- **๐งช Tests**: Maintain 100% test coverage
- **๐จ Code Style**: Follow Ruff formatting and type hints
### ๐๏ธ Development Workflow
1. **Fork** the repository
2. **Create** a feature branch: `git checkout -b feature/amazing-feature`
3. **Make** your changes with tests
4. **Run** the test suite: `pytest --cov=fastrict`
5. **Commit** with clear messages: `git commit -m 'Add amazing feature'`
6. **Push** to your fork: `git push origin feature/amazing-feature`
7. **Create** a Pull Request
## ๐ Resources & Documentation
### ๐ Documentation
- **[API Reference](https://github.com/msameim181/fastrict)** - Complete API documentation
- **[User Guide](https://github.com/msameim181/fastrict)** - Step-by-step tutorials
- **[Examples](https://github.com/msameim181/fastrict/tree/main/examples)** - Real-world examples
- **[Architecture](https://github.com/msameim181/fastrict)** - Design decisions
### ๐ Support Channels
- **๐ [Issue Tracker](https://github.com/msameim181/fastrict/issues)** - Bug reports & feature requests
- **๐ฌ [Discussions](https://github.com/msameim181/fastrict/discussions)** - Community Q&A
- **๐ง [Email](mailto:9259samei@gmail.com)** - Direct support for enterprise users
- **๐ผ [LinkedIn](https://linkedin.com/in/msameim181)** - Professional inquiries
### ๐ Related Projects
- **[FastAPI](https://fastapi.tiangolo.com/)** - Modern, fast web framework for building APIs
- **[Redis](https://redis.io/)** - In-memory data structure store
- **[Starlette](https://www.starlette.io/)** - Lightweight ASGI framework
- **[Pydantic](https://pydantic-docs.helpmanual.io/)** - Data validation using Python type hints
## ๐ License
This project is licensed under the **MIT License** - see the [LICENSE](LICENSE) file for details.
## ๐ Changelog & Roadmap
### ๐ฏ Current Version: `v0.1.1`
See [CHANGELOG.md](CHANGELOG.md) for version history and release notes.
### ๐ Upcoming Features
- **๐ GraphQL Support**: Rate limiting for GraphQL endpoints
- **๐ Django Support**: Rate limiting for Django applications
- **๐ Prometheus Metrics**: Built-in metrics collection
- **๐ Circuit Breaker**: Integrate with circuit breaker patterns
- **๐ฏ Rate Limit Warming**: Gradual limit increases
- **๐ฑ WebSocket Support**: Rate limiting for WebSocket connections
---
<div align="center">
**Fastrict - Powering the next generation of FastAPI applications**
[โฌ๏ธ Back to Top](#-fastrict---enterprise-fastapi-rate-limiter)
</div>
Raw data
{
"_id": null,
"home_page": "https://github.com/msameim181/fastrict",
"name": "fastrict",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "fastapi, rate limiting, redis, middleware, decorator, throttle",
"author": "Mohammad Mahdi Samei",
"author_email": "Mohammad Mahdi Samei <9259samei@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/d3/4f/53a9d2bcb5c9a36dc275251ca11cb012e4a0621109a9ff0bd585e1f285da/fastrict-0.1.2.tar.gz",
"platform": null,
"description": "# \ud83d\ude80 Fastrict - Enterprise FastAPI Rate Limiter\n\n**The most powerful, flexible, and production-ready rate limiting system for FastAPI applications.**\n\nFastrict provides enterprise-grade rate limiting with Redis and in-memory backends, supporting everything from simple API throttling to complex multi-tenant rate limiting strategies.\n\n[](https://www.python.org/downloads/)\n[](https://fastapi.tiangolo.com/)\n[](https://redis.io/)\n[](https://opensource.org/licenses/MIT)\n[](https://pypi.org/project/fastrict/)\n[](https://pypi.org/project/fastrict/)\n[](#-performance-benchmarks)\n[](#-performance-benchmarks)\n\n## \u2728 Features\n\n### \ud83c\udfd7\ufe0f **Dual Architecture Support**\n- **\ud83c\udf10 Global Rate Limiting**: Shared limits across all endpoints\n- **\ud83c\udfaf Per-Route Rate Limiting**: Independent limits for each endpoint\n- **\ud83d\udd04 Hybrid Mode**: Mix global and per-route limits in the same application\n\n### \ud83d\ude80 **Extreme Performance**\n- **\u26a1 Sub-millisecond latency**: Ultra-fast rate limit checks\n- **\ud83d\udcca 1K-30K concurrent connections**: Enterprise-scale performance\n- **\ud83e\uddee Sliding window algorithm**: Precise rate limiting with Redis sorted sets\n- **\ud83d\uddd1\ufe0f Automatic cleanup**: Expired keys removed automatically\n\n### \ufffd **Advanced Key Extraction**\n- **\ud83c\udf0d IP-based limiting**: Traditional client IP throttling\n- **\ud83d\udd11 Header-based**: API keys, user tokens, custom headers\n- **\ud83d\udccb Query parameters**: Rate limit by user ID, tenant, etc.\n- **\ud83d\udcdd Form fields**: POST form data extraction\n- **\ud83c\udfad Custom functions**: Complex business logic extraction\n- **\ud83d\udd17 Combined keys**: Multi-factor rate limiting (IP + API key + tenant)\n\n### \ufffd\ufe0f **Intelligent Bypass System**\n- **\ud83d\udc51 Role-based bypass**: Skip limits for admin users\n- **\ud83c\udfab Premium tier bypass**: Different limits for paid users\n- **\ud83d\udd27 Maintenance mode**: Conditional bypass during deployments\n- **\ud83e\udd16 Custom logic**: Any business rule for bypass decisions\n\n### \ud83d\udcca **Production Monitoring**\n- **\ud83d\udcc8 Standard HTTP headers**: `X-RateLimit-*` headers\n- **\ud83d\udcf1 Real-time usage**: Current count, remaining, usage percentage\n- **\u23f1\ufe0f Retry-After**: Smart retry timing\n- **\ud83d\udccb Comprehensive logging**: Structured logs for monitoring\n- **\ud83c\udfaf Usage statistics**: Track rate limit effectiveness\n\n### \ud83c\udfed **Enterprise Ready**\n- **\u2601\ufe0f Redis Cluster support**: Horizontal scaling\n- **\ud83d\udcbe Memory fallback**: In-memory storage for development\n- **\ud83d\udd04 Graceful degradation**: Continues working if Redis fails\n- **\ud83d\udd12 Thread-safe**: Concurrent request handling\n- **\ud83e\uddea 100% test coverage**: Thoroughly tested codebase\n- **\ud83d\udccb Clean Architecture**: SOLID principles, easy to extend\n\n## \ud83d\udce6 Installation\n\n```bash\n# Install from PyPI\npip install fastrict\n\n# Install with development dependencies\npip install fastrict[dev]\n\n# Install with documentation dependencies\npip install fastrict[docs]\n```\n\n### \ud83d\udd27 System Requirements\n\n| Component | Version | Purpose |\n|-----------|---------|---------|\n| **Python** | 3.8+ | Core runtime |\n| **FastAPI** | 0.68+ | Web framework |\n| **Redis** | 4.0+ | Primary storage backend |\n| **Pydantic** | 1.8+ | Data validation |\n| **Starlette** | 0.14+ | ASGI framework |\n\n## \ud83d\ude80 Quick Start\n\n### \ud83c\udfaf 1. Basic Setup (30 seconds)\n\n```python\nfrom fastapi import FastAPI\nfrom fastrict import RateLimitMiddleware, RedisRateLimitRepository\nfrom fastrict import RateLimitUseCase, KeyExtractionUseCase\n\n# Create FastAPI app\napp = FastAPI(title=\"My Rate Limited API\")\n\n# Setup rate limiting (Redis)\nrepository = RedisRateLimitRepository.from_url(\"redis://localhost:6379\")\nkey_extraction = KeyExtractionUseCase()\nrate_limiter = RateLimitUseCase(repository, key_extraction)\n\n# Create default key extraction strategy (NEW in v0.1.0)\n# Try API key, then Authorization header, then fall back to IP\nfrom fastrict import create_api_key_fallback\ndefault_key_extraction = create_api_key_fallback()\n\n# Add global rate limiting middleware\napp.add_middleware(\n RateLimitMiddleware,\n rate_limit_use_case=rate_limiter,\n excluded_paths=[\"/health\", \"/docs\", \"/metrics\"],\n default_key_extraction=default_key_extraction # NEW: Default for all routes\n)\n\n@app.get(\"/api/data\")\nasync def get_data():\n return {\"message\": \"This endpoint is globally rate limited\"}\n```\n\n### \ud83c\udfa8 2. Route-Specific Rate Limiting\n\n```python\nfrom fastrict import throttle, RateLimitStrategyName, RateLimitMode\n\n# Strict rate limiting for authentication\n@app.post(\"/auth/login\")\n@throttle(strategy=RateLimitStrategyName.SHORT) # 3 requests per minute\nasync def login():\n return {\"token\": \"jwt-token-here\"}\n\n# Custom rate limiting for file uploads\n@app.post(\"/api/upload\")\n@throttle(limit=5, ttl=300) # 5 uploads per 5 minutes\nasync def upload_file():\n return {\"file_id\": \"12345\", \"status\": \"uploaded\"}\n\n# Premium endpoint with generous limits\n@app.get(\"/api/premium\")\n@throttle(limit=1000, ttl=3600) # 1000 requests per hour\nasync def premium_data():\n return {\"data\": \"premium content\"}\n```\n\n### \ud83d\udd11 3. Advanced Key Extraction\n\n```python\nfrom fastrict import KeyExtractionType\n\n# API key-based rate limiting\n@app.get(\"/api/protected\")\n@throttle(\n limit=100, \n ttl=3600,\n key_type=KeyExtractionType.HEADER,\n key_field=\"X-API-Key\",\n key_default=\"anonymous\"\n)\nasync def protected_endpoint():\n return {\"data\": \"API key limited content\"}\n\n# User-specific rate limiting\n@app.get(\"/api/user-data\")\n@throttle(\n limit=50,\n ttl=600,\n key_type=KeyExtractionType.QUERY_PARAM,\n key_field=\"user_id\",\n key_default=\"guest\"\n)\nasync def user_data():\n return {\"data\": \"user-specific data\"}\n\n# Multi-factor rate limiting (IP + API key)\n@app.get(\"/api/sensitive\")\n@throttle(\n limit=10,\n ttl=300,\n key_type=KeyExtractionType.COMBINED,\n key_combination=[\"ip\", \"header:X-API-Key\"]\n)\nasync def sensitive_data():\n return {\"data\": \"highly sensitive information\"}\n```\n\n## \ud83c\udf9b\ufe0f Rate Limiting Modes\n\nFastrict offers two powerful rate limiting modes that can be mixed and matched:\n\n### \ud83c\udf10 Global Mode\nAll endpoints share the same rate limit pool. Perfect for overall API protection.\n\n```python\nfrom fastrict import RateLimitMode\n\napp.add_middleware(\n RateLimitMiddleware,\n rate_limit_use_case=rate_limiter,\n rate_limit_mode=RateLimitMode.GLOBAL, # All routes share limits\n default_strategy_name=RateLimitStrategyName.MEDIUM\n)\n\n@app.get(\"/api/data\") # \u2500\u2500\u2510 \nasync def get_data(): # \u251c\u2500\u2500 All share same\n return {\"data\": \"...\"} # \u2502 20 req/10min pool\n\n@app.get(\"/api/users\") # \u2502\nasync def get_users(): # \u2502 \n return {\"users\": []} # \u2500\u2500\u2518\n```\n\n### \ud83c\udfaf Per-Route Mode\nEach endpoint has independent rate limit pools. Ideal for fine-grained control.\n\n```python\napp.add_middleware(\n RateLimitMiddleware,\n rate_limit_use_case=rate_limiter,\n rate_limit_mode=RateLimitMode.PER_ROUTE # Independent limits per route\n)\n\n@app.get(\"/api/data\") # \u2500\u2500 20 req/10min (independent)\nasync def get_data():\n return {\"data\": \"...\"}\n\n@app.get(\"/api/users\") # \u2500\u2500 20 req/10min (independent) \nasync def get_users():\n return {\"users\": []}\n```\n\n### \ud83d\udd04 Hybrid Mode\nMix global middleware with per-route decorators for ultimate flexibility:\n\n```python\n# Global middleware (GLOBAL mode)\napp.add_middleware(\n RateLimitMiddleware,\n rate_limit_use_case=rate_limiter,\n rate_limit_mode=RateLimitMode.GLOBAL\n)\n\n@app.get(\"/api/public\") # Uses global pool\nasync def public_data():\n return {\"data\": \"public\"}\n\n@app.get(\"/api/special\") # Gets its own independent pool\n@throttle(limit=100, ttl=3600, rate_limit_mode=RateLimitMode.PER_ROUTE)\nasync def special_endpoint():\n return {\"data\": \"special\"}\n```\n\n## \ufffd Fallback Key Extraction Strategies\n\n**NEW in v0.1.0**: Advanced fallback mechanisms that try multiple extraction methods in sequence.\n\n### \ud83c\udfd7\ufe0f Built-in Fallback Helpers\n\nFastrict provides convenient helper functions for common fallback patterns:\n\n```python\nfrom fastrict import (\n create_auth_header_fallback,\n create_api_key_fallback, \n create_user_id_fallback\n)\n\n# Try Authorization header, then IP\nauth_fallback = create_auth_header_fallback(\n header_name=\"Authorization\", # Default\n default_value=\"anonymous\" # Optional\n)\n\n# Try API key, then Authorization, then IP\napi_fallback = create_api_key_fallback(\n api_key_header=\"X-API-Key\", # Default \n auth_header=\"Authorization\", # Default\n default_value=None # Will use IP if headers missing\n)\n\n# Try user ID from query param, then header, then IP\nuser_fallback = create_user_id_fallback(\n user_id_param=\"user_id\", # Default\n user_id_header=\"X-User-ID\", # Default\n default_value=\"anonymous\" # Optional\n)\n```\n\n### \u2699\ufe0f Middleware Default Strategy\n\nSet a default key extraction strategy that applies to all routes:\n\n```python\nfrom fastrict import RateLimitMiddleware, create_api_key_fallback\n\n# Create fallback strategy for middleware\ndefault_strategy = create_api_key_fallback(\n api_key_header=\"X-API-Key\",\n auth_header=\"Authorization\"\n # Falls back to IP if neither header is present\n)\n\napp.add_middleware(\n RateLimitMiddleware,\n rate_limit_use_case=rate_limiter,\n default_key_extraction=default_strategy, # Applied to all routes\n rate_limit_mode=RateLimitMode.GLOBAL\n)\n\n# This endpoint will use the middleware default strategy\n@app.get(\"/api/data\")\nasync def get_data():\n return {\"data\": \"Uses API key \u2192 Auth header \u2192 IP fallback\"}\n\n# This endpoint overrides with its own strategy \n@app.get(\"/api/users\")\n@throttle(\n limit=50, ttl=3600,\n key_extraction_strategy=create_user_id_fallback()\n)\nasync def get_users():\n return {\"users\": \"Uses user ID \u2192 header \u2192 IP fallback\"}\n```\n\n### \ud83c\udfaf Route-Specific Fallback\n\nOverride the middleware default for specific routes:\n\n```python\n# Use helper function directly\n@app.get(\"/api/auth-required\")\n@throttle(\n limit=100, ttl=3600,\n key_extraction_strategy=create_auth_header_fallback()\n)\nasync def auth_endpoint():\n return {\"data\": \"auth-protected\"}\n\n# Custom fallback strategy\nfrom fastrict import KeyExtractionStrategy, KeyExtractionType\n\ncustom_fallback = KeyExtractionStrategy(\n type=KeyExtractionType.FALLBACK,\n fallback_strategies=[\n KeyExtractionStrategy(\n type=KeyExtractionType.HEADER,\n field_name=\"X-Session-ID\"\n ),\n KeyExtractionStrategy(\n type=KeyExtractionType.HEADER, \n field_name=\"X-API-Key\"\n ),\n KeyExtractionStrategy(\n type=KeyExtractionType.IP\n )\n ]\n)\n\n@app.get(\"/api/session-data\")\n@throttle(\n limit=50, ttl=600,\n key_extraction_strategy=custom_fallback\n)\nasync def session_endpoint():\n return {\"data\": \"session-based rate limiting\"}\n```\n\n### \ud83d\udd04 How Fallback Works\n\n1. **Try first strategy**: Attempt to extract key using the first method\n2. **Check success**: If extraction succeeds and returns a valid key, use it\n3. **Try next strategy**: If extraction fails or returns empty, try next method\n4. **Continue sequence**: Repeat until a strategy succeeds\n5. **IP fallback**: If all strategies fail, fall back to IP address\n\n```python\n# Example: API key \u2192 Auth header \u2192 IP fallback\napi_fallback = create_api_key_fallback()\n\n# For a request with these headers:\n# X-API-Key: \"\" (empty)\n# Authorization: \"Bearer token123\"\n# Client IP: \"192.168.1.100\"\n\n# Fallback process:\n# 1. Try X-API-Key \u2192 empty, skip\n# 2. Try Authorization \u2192 \"Bearer token123\" \u2713\n# Result: Rate limiting key = \"Bearer token123\"\n```\n\n### \ud83c\udfe2 Real-World Example\n\n```python\n# Multi-tenant SaaS with intelligent key extraction\nfrom fastrict import create_api_key_fallback, RateLimitMode\n\n# Middleware default: API key for tenant isolation\ndefault_strategy = create_api_key_fallback(\n api_key_header=\"X-API-Key\",\n auth_header=\"Authorization\"\n)\n\napp.add_middleware(\n RateLimitMiddleware,\n rate_limit_use_case=rate_limiter,\n default_key_extraction=default_strategy,\n rate_limit_mode=RateLimitMode.GLOBAL,\n default_strategy_name=RateLimitStrategyName.MEDIUM\n)\n\n# Public endpoints use IP-based limiting\n@app.get(\"/api/public\")\n@throttle(\n limit=100, ttl=3600,\n key_extraction_strategy=KeyExtractionStrategy(type=KeyExtractionType.IP)\n)\nasync def public_data():\n return {\"data\": \"public\"}\n\n# User endpoints prefer user ID over API key\n@app.get(\"/api/user-profile\")\n@throttle(\n limit=200, ttl=3600,\n key_extraction_strategy=create_user_id_fallback()\n)\nasync def user_profile():\n return {\"profile\": \"user data\"}\n\n# Admin endpoints use session-based limiting\nadmin_fallback = KeyExtractionStrategy(\n type=KeyExtractionType.FALLBACK,\n fallback_strategies=[\n KeyExtractionStrategy(type=KeyExtractionType.HEADER, field_name=\"Admin-Session\"),\n KeyExtractionStrategy(type=KeyExtractionType.HEADER, field_name=\"X-API-Key\"),\n KeyExtractionStrategy(type=KeyExtractionType.IP)\n ]\n)\n\n@app.get(\"/api/admin\")\n@throttle(\n limit=1000, ttl=3600,\n key_extraction_strategy=admin_fallback\n)\nasync def admin_endpoint():\n return {\"data\": \"admin-only\"}\n```\n\n## \ufffd\ud83d\udd11 Key Extraction Strategies\n\n### \ud83d\udccd IP-Based (Default)\n```python\n@throttle(limit=100, ttl=3600) # Rate limit per client IP\n```\n\n### \ud83c\udfab Header-Based\n```python\n# API key rate limiting\n@throttle(\n limit=1000, ttl=3600,\n key_type=KeyExtractionType.HEADER,\n key_field=\"X-API-Key\",\n key_default=\"anonymous\"\n)\n\n# User token rate limiting \n@throttle(\n limit=500, ttl=3600,\n key_type=KeyExtractionType.HEADER,\n key_field=\"Authorization\",\n key_default=\"unauthenticated\"\n)\n```\n\n### \ud83d\udccb Query Parameter-Based\n```python\n# User-specific limits\n@throttle(\n limit=200, ttl=3600,\n key_type=KeyExtractionType.QUERY_PARAM,\n key_field=\"user_id\",\n key_default=\"anonymous\"\n)\n\n# Tenant-based limits (SaaS)\n@throttle(\n limit=10000, ttl=3600,\n key_type=KeyExtractionType.QUERY_PARAM, \n key_field=\"tenant_id\",\n key_default=\"free_tier\"\n)\n```\n\n### \ud83d\udd17 Combined Key Strategies\n```python\n# Multi-factor rate limiting\n@throttle(\n limit=50, ttl=300,\n key_type=KeyExtractionType.COMBINED,\n key_combination=[\n \"ip\", # Client IP\n \"header:X-API-Key\", # API key\n \"query_param:tenant_id\" # Tenant\n ]\n)\n# Results in key: \"192.168.1.1:abc123:tenant_456\"\n```\n\n### \ud83c\udfad Custom Key Extraction\n```python\ndef extract_session_key(request: Request) -> str:\n \"\"\"Complex business logic for key extraction.\"\"\"\n session_id = request.headers.get(\"Session-ID\")\n user_tier = request.headers.get(\"User-Tier\", \"free\")\n \n if user_tier == \"premium\":\n return f\"premium:session:{session_id}\"\n elif user_tier == \"enterprise\":\n return f\"enterprise:session:{session_id}\"\n else:\n return f\"free:ip:{request.client.host}\"\n\n@throttle(\n limit=100, ttl=3600,\n key_type=KeyExtractionType.CUSTOM,\n key_extractor=extract_session_key\n)\nasync def complex_endpoint():\n return {\"data\": \"complex rate limiting\"}\n```\n\n## \ud83d\udee1\ufe0f Smart Bypass System\n\nCreate intelligent bypass rules for different user roles, maintenance modes, or business logic.\n\n### \ud83d\udc51 Role-Based Bypass\n```python\ndef bypass_for_admins(request: Request) -> bool:\n \"\"\"Bypass rate limiting for admin users.\"\"\"\n user_role = request.headers.get(\"User-Role\")\n return user_role in [\"admin\", \"superuser\"]\n\n@app.get(\"/api/admin-only\")\n@throttle(\n limit=10, ttl=60,\n bypass_function=bypass_for_admins,\n custom_error_message=\"Admin endpoint requires admin privileges\"\n)\nasync def admin_endpoint():\n return {\"data\": \"admin-only data\"}\n```\n\n### \ud83c\udfab Premium User Bypass\n```python\ndef bypass_for_premium(request: Request) -> bool:\n \"\"\"Bypass limits for premium subscribers.\"\"\"\n subscription = request.headers.get(\"Subscription-Tier\")\n return subscription in [\"premium\", \"enterprise\"]\n\n@app.get(\"/api/premium-features\")\n@throttle(\n limit=5, ttl=60, # Limits for free users\n bypass_function=bypass_for_premium\n)\nasync def premium_features():\n return {\"features\": [\"advanced\", \"priority\"]}\n```\n\n### \ud83d\udd27 Maintenance Mode Bypass\n```python\nimport os\n\ndef bypass_during_maintenance(request: Request) -> bool:\n \"\"\"Bypass rate limiting during maintenance.\"\"\"\n maintenance_mode = os.getenv(\"MAINTENANCE_MODE\", \"false\").lower() == \"true\"\n maintenance_key = request.headers.get(\"Maintenance-Key\")\n \n return maintenance_mode and maintenance_key == os.getenv(\"MAINTENANCE_SECRET\")\n\n@app.get(\"/api/critical\")\n@throttle(\n limit=100, ttl=3600,\n bypass_function=bypass_during_maintenance\n)\nasync def critical_endpoint():\n return {\"data\": \"critical system data\"}\n```\n\n## \ud83d\udcca Built-in Strategies\n\nFastrict comes with pre-configured strategies for common use cases:\n\n```python\nfrom fastrict import RateLimitStrategy, RateLimitStrategyName\n\n# Define custom strategies\ncustom_strategies = [\n RateLimitStrategy(\n name=RateLimitStrategyName.SHORT, \n limit=3, \n ttl=60\n ), # Strict: 3 requests per minute\n \n RateLimitStrategy(\n name=RateLimitStrategyName.MEDIUM, \n limit=20, \n ttl=600\n ), # Moderate: 20 requests per 10 minutes\n \n RateLimitStrategy(\n name=RateLimitStrategyName.LONG, \n limit=100, \n ttl=3600\n ), # Generous: 100 requests per hour\n]\n\napp.add_middleware(\n RateLimitMiddleware,\n rate_limit_use_case=rate_limiter,\n default_strategies=custom_strategies,\n default_strategy_name=RateLimitStrategyName.MEDIUM\n)\n\n# Use predefined strategies\n@app.post(\"/auth/login\")\n@throttle(strategy=RateLimitStrategyName.SHORT) # Use strict limits\nasync def login():\n return {\"message\": \"Login attempt\"}\n\n@app.get(\"/api/search\") \n@throttle(strategy=RateLimitStrategyName.LONG) # Use generous limits\nasync def search():\n return {\"results\": []}\n```\n\n## \ud83c\udfd7\ufe0f Storage Backends\n\n### \u26a1 Redis Backend (Recommended)\nPerfect for production, supports clustering and persistence.\n\n```python\nfrom fastrict import RedisRateLimitRepository\n\n# Simple connection\nrepository = RedisRateLimitRepository.from_url(\"redis://localhost:6379\")\n\n# Advanced configuration\nrepository = RedisRateLimitRepository.from_url(\n redis_url=\"redis://:password@localhost:6379/0\",\n key_prefix=\"myapp_limits\",\n logger=my_logger\n)\n\n# Custom Redis client\nimport redis\nredis_client = redis.Redis(\n host=\"localhost\",\n port=6379,\n password=\"secret\",\n decode_responses=True,\n socket_timeout=5,\n retry_on_timeout=True\n)\nrepository = RedisRateLimitRepository(\n redis_client=redis_client,\n key_prefix=\"production_limits\"\n)\n```\n\n### \ud83d\udcbe Memory Backend (Development)\nGreat for testing and development environments.\n\n```python\nfrom fastrict import MemoryRateLimitRepository\n\n# In-memory storage (no persistence)\nrepository = MemoryRateLimitRepository(\n key_prefix=\"dev_limits\",\n cleanup_interval=300 # Cleanup every 5 minutes\n)\n```\n\n## \ud83d\udcca Monitoring & Observability\n\n### \ud83d\udcc8 Standard HTTP Headers\nFastrict automatically adds industry-standard rate limiting headers:\n\n```http\nHTTP/1.1 200 OK\nX-RateLimit-Limit: 100 # Maximum requests in window\nX-RateLimit-Remaining: 75 # Requests remaining in window \nX-RateLimit-Used: 25 # Requests used in window\nX-RateLimit-Window: 3600 # Window duration in seconds\n```\n\nWhen rate limited (HTTP 429):\n```http\nHTTP/1.1 429 Too Many Requests\nX-RateLimit-Limit: 100\nX-RateLimit-Remaining: 0\nX-RateLimit-Used: 100\nX-RateLimit-Window: 3600\nRetry-After: 1847 # Seconds until window resets\n```\n\n### \ud83d\udcf1 Real-time Status Endpoint\n```python\n@app.get(\"/api/rate-limit-status\")\n@throttle(bypass=True) # Don't count status checks against limits\nasync def rate_limit_status(request: Request):\n \"\"\"Get current rate limit status without incrementing counter.\"\"\"\n result = rate_limiter.get_current_usage(\n request=request,\n middleware_rate_limit_mode=RateLimitMode.GLOBAL,\n route_path=request.url.path\n )\n \n return {\n \"allowed\": result.allowed,\n \"current_count\": result.current_count,\n \"limit\": result.limit,\n \"remaining\": result.remaining_requests,\n \"reset_in_seconds\": result.ttl,\n \"usage_percentage\": result.usage_percentage,\n \"strategy\": result.strategy_name,\n \"key\": result.key # Rate limiting key used\n }\n```\n\n### \ud83d\udccb Structured Error Responses\n```json\n{\n \"message\": \"Rate limit exceeded. Maximum 100 requests per 3600 seconds. Please try again in 1847 seconds.\",\n \"retry_after\": 1847,\n \"limit\": 100,\n \"window\": 3600,\n \"current_count\": 100,\n \"usage_percentage\": 100.0,\n \"strategy\": \"medium\"\n}\n```\n\n### \ud83d\udd27 Custom Error Messages\n```python\n@app.post(\"/api/critical\")\n@throttle(\n limit=5, ttl=60,\n custom_error_message=\"Critical endpoint allows only 5 requests per minute. Please use batch operations for bulk requests.\"\n)\nasync def critical_operation():\n return {\"status\": \"processing\"}\n```\n\n## \ud83e\uddea Testing Your Rate Limits\n\n### \ud83d\udcdd Unit Testing\n```python\nimport pytest\nfrom fastapi.testclient import TestClient\nfrom unittest.mock import Mock\n\ndef test_rate_limiting():\n # Mock Redis for testing\n mock_redis = Mock()\n repository = RedisRateLimitRepository(mock_redis)\n \n with TestClient(app) as client:\n # First request should succeed\n response = client.get(\"/api/data\")\n assert response.status_code == 200\n assert \"X-RateLimit-Remaining\" in response.headers\n \n # Simulate rate limit exceeded\n mock_redis.zcard.return_value = 100 # Over limit\n response = client.get(\"/api/data\")\n assert response.status_code == 429\n assert \"Retry-After\" in response.headers\n```\n\n### \ud83d\udd04 Integration Testing\n```python\nimport asyncio\nimport httpx\n\nasync def test_concurrent_requests():\n \"\"\"Test rate limiting under concurrent load.\"\"\"\n async with httpx.AsyncClient() as client:\n # Fire 10 concurrent requests\n tasks = [\n client.get(\"http://localhost:8000/api/data\")\n for _ in range(10)\n ]\n responses = await asyncio.gather(*tasks)\n \n # Check that some are rate limited\n success_count = sum(1 for r in responses if r.status_code == 200)\n rate_limited_count = sum(1 for r in responses if r.status_code == 429)\n \n assert success_count <= 5 # Our test limit\n assert rate_limited_count >= 5\n```\n\n### \ud83d\udea8 Load Testing\n```bash\n# Install hey for load testing\ngo install github.com/rakyll/hey@latest\n\n# Test rate limiting under load\nhey -n 100 -c 10 -H \"X-API-Key: test123\" http://localhost:8000/api/data\n\n# Expected output shows rate limiting in action:\n# Status code distribution:\n# [200] 20 responses (successful requests)\n# [429] 80 responses (rate limited)\n```\n\n## \ud83d\ude80 Performance Characteristics\n\n### \u26a1 Benchmarks\n\n| Metric | Value | Notes |\n|--------|-------|-------|\n| **Latency** | < 1ms | Rate limit check overhead |\n| **Throughput** | 30K+ req/s | Redis backend, single instance |\n| **Memory** | ~10MB | Per 100K active keys |\n| **CPU** | < 1% | Minimal overhead |\n\n### \ud83d\udcca Scalability\n\n```python\n# Horizontal scaling with Redis Cluster\nrepository = RedisRateLimitRepository.from_url(\n \"redis://node1:7000,node2:7000,node3:7000\",\n key_prefix=\"cluster_limits\"\n)\n\n# Multiple app instances can share rate limits\n# Perfect for microservices and load-balanced deployments\n```\n\n## \ud83c\udfd7\ufe0f Architecture & Design\n\nFastrict follows **Clean Architecture** principles:\n\n```\nsrc/fastrict/\n\u251c\u2500\u2500 entities/ # \ud83c\udfdb\ufe0f Core business models & enums\n\u2502 \u251c\u2500\u2500 models.py # RateLimitStrategy, RateLimitResult\n\u2502 \u2514\u2500\u2500 enums.py # KeyExtractionType, RateLimitMode\n\u251c\u2500\u2500 use_cases/ # \ud83e\udde0 Business logic & orchestration \n\u2502 \u251c\u2500\u2500 rate_limit.py # Core rate limiting logic\n\u2502 \u2514\u2500\u2500 key_extraction.py Key extraction strategies\n\u251c\u2500\u2500 adapters/ # \ud83d\udd0c External integrations\n\u2502 \u251c\u2500\u2500 redis_repository.py Redis storage backend\n\u2502 \u2514\u2500\u2500 memory_repository.py In-memory storage backend \n\u2514\u2500\u2500 frameworks/ # \ud83c\udf10 FastAPI integration\n \u251c\u2500\u2500 middleware.py # Global rate limiting middleware\n \u2514\u2500\u2500 decorator.py # @throttle route decorator\n```\n\n### \ud83c\udfaf Design Principles\n\n- **\ud83d\udd12 Immutable Entities**: Thread-safe by design\n- **\ud83e\uddea Dependency Injection**: Easy testing and mocking\n- **\ud83d\udd0c Interface Segregation**: Swap backends seamlessly \n- **\ud83d\udce6 Single Responsibility**: Each component has one job\n- **\ud83d\ude80 Performance First**: Optimized for high throughput\n\n## \ud83d\udcca Performance Benchmarks\n\n*Last updated: 2025-10-02 (MacOS 26, M3 Pro, conda chat environment)*\n\nFastrict has been extensively tested for performance under various load conditions. Here are the benchmark results:\n\n### \u26a1 Single Request Performance\n\n| Metric | Value | Description |\n|--------|-------|-------------|\n| **Single Request Latency** | **0.37 ms** | Ultra-fast rate limit check overhead |\n\n### \ud83c\udfc3\u200d\u2642\ufe0f Sequential Performance\n\n| Metric | Value | Description |\n|--------|-------|-------------|\n| **Total Requests** | 1,000 | Sequential test requests |\n| **Duration** | 0.35 seconds | Total test time |\n| **Requests/Second** | **2,857 RPS** | Sequential throughput |\n| **Average Response Time** | **0.35 ms** | Mean response time |\n| **P95 Response Time** | **0.41 ms** | 95th percentile |\n\n### \ud83d\ude80 Concurrent Performance (High Load)\n\n| Metric | Value | Description |\n|--------|-------|-------------|\n| **Total Requests** | 1,000 | 50 users \u00d7 20 requests each |\n| **Duration** | 0.27 seconds | Concurrent execution time |\n| **Requests/Second** | **3,676 RPS** | Concurrent throughput |\n| **Success Rate** | **100.0%** | Zero failures under load |\n| **Average Response Time** | **13.41 ms** | Mean response time |\n| **P95 Response Time** | **28.64 ms** | 95th percentile |\n| **P99 Response Time** | **28.93 ms** | 99th percentile |\n\n### \ud83d\udee1\ufe0f Rate Limiting Accuracy\n\n| Metric | Value | Description |\n|--------|-------|-------------|\n| **Total Requests** | 100 | Concurrent requests to limited endpoint |\n| **Successful Requests** | 50 | Requests within limit |\n| **Rate Limited Requests** | 50 | Correctly blocked requests |\n| **Accuracy** | **100%** | Perfect rate limiting enforcement |\n| **Average Response Time** | **10.37 ms** | Fast even when blocking |\n\n### \ud83d\udcaa Extreme Load Test\n\n| Metric | Value | Description |\n|--------|-------|-------------|\n| **Total Requests** | 1,000 | 100 users \u00d7 10 requests each |\n| **Requests/Second** | **3,639 RPS** | Sustained under extreme load |\n| **Success Rate** | **100.0%** | No failures under pressure |\n| **Error Rate** | **0.0%** | System stability maintained |\n| **P99 Response Time** | **32.56 ms** | Excellent tail latency |\n\n### \ud83d\udd04 Sustained Load Endurance\n\n| Metric | Value | Description |\n|--------|-------|-------------|\n| **Total Requests** | 913 | 10-second endurance test |\n| **Achieved RPS** | **91.24** | Target: 100 RPS |\n| **Success Rate** | **100.0%** | No degradation over time |\n| **Average Response Time** | **1.95 ms** | Consistent performance |\n| **Performance Degradation** | **21.7%** | Minimal performance loss |\n\n### \ud83c\udfc6 Performance Highlights\n\n- \u26a1 **Sub-millisecond latency**: 0.37ms average response time\n- \ud83d\ude80 **3,600+ RPS**: Exceptional concurrent throughput\n- \ud83c\udfaf **100% success rate**: Perfect stability under load\n- \ud83d\udee1\ufe0f **100% rate limiting accuracy**: Precise enforcement\n- \ud83d\udcbe **Memory efficient**: Handles thousands of unique keys\n- \ud83d\udd04 **Minimal degradation**: Stable performance over time\n\n### \ud83e\uddea Test Environment\n\n- **Hardware**: MacOS, M1 Pro\n- **Python**: 3.10.16 (conda environment)\n- **Backend**: In-memory storage (optimal performance)\n- **Test Framework**: pytest + httpx + asyncio\n- **Load Patterns**: Sequential, concurrent, sustained, extreme scenarios\n\n### \ud83d\udd2c Run Performance Tests Yourself\n\nWant to verify these results? Run the performance tests on your own system:\n\n```bash\n# Install dependencies\nconda activate chat # or your preferred environment\npip install pytest httpx pytest-asyncio uvicorn\npip install -e .\n\n# Run comprehensive performance test suite\npython -m pytest tests/test_performance.py -v\n\n# Run live performance demo\npython test/demo_performance.py\n\n# Generate performance report\npython test/run_performance_tests.py\n```\n\nSee [`PERFORMANCE_SUMMARY.md`](PERFORMANCE_SUMMARY.md) and [`tests/PERFORMANCE.md`](tests/PERFORMANCE.md) for detailed testing documentation.\n\n### \ud83d\ude80 Real-World Performance\n\nThese benchmarks demonstrate that Fastrict can easily handle:\n\n- **High-traffic APIs**: 3,000+ requests per second\n- **Real-time applications**: Sub-millisecond response times\n- **Microservices**: Zero performance impact\n- **Enterprise workloads**: 100% stability under pressure\n\n*Performance may vary based on hardware, Redis configuration, and network conditions.*\n\n## \ud83c\udfaf Real-World Examples\n\n### \ud83c\udfe2 Multi-Tenant SaaS Application\n```python\ndef extract_tenant_key(request: Request) -> str:\n \"\"\"Extract tenant-aware rate limiting key.\"\"\"\n api_key = request.headers.get(\"X-API-Key\", \"\")\n tenant_id = request.headers.get(\"X-Tenant-ID\", \"unknown\")\n \n # Different limits based on subscription tier\n if api_key.startswith(\"ent_\"):\n return f\"enterprise:tenant:{tenant_id}\"\n elif api_key.startswith(\"pro_\"):\n return f\"professional:tenant:{tenant_id}\"\n else:\n return f\"free:tenant:{tenant_id}\"\n\n# Different strategies per tier\nenterprise_strategy = RateLimitStrategy(name=RateLimitStrategyName.CUSTOM, limit=10000, ttl=3600)\nprofessional_strategy = RateLimitStrategy(name=RateLimitStrategyName.LONG, limit=1000, ttl=3600)\nfree_strategy = RateLimitStrategy(name=RateLimitStrategyName.MEDIUM, limit=100, ttl=3600)\n\n@app.get(\"/api/analytics\")\n@throttle(\n limit=100, # Free tier limit\n ttl=3600,\n key_type=KeyExtractionType.CUSTOM,\n key_extractor=extract_tenant_key\n)\nasync def get_analytics():\n return {\"analytics\": \"tenant-specific data\"}\n```\n\n### \ud83d\uded2 E-commerce API Protection \n```python\n# Protect checkout process\n@app.post(\"/api/checkout\")\n@throttle(\n limit=5, ttl=300, # 5 checkouts per 5 minutes\n key_type=KeyExtractionType.HEADER,\n key_field=\"User-ID\",\n custom_error_message=\"Too many checkout attempts. Please wait before trying again.\"\n)\nasync def process_checkout():\n return {\"order_id\": \"12345\", \"status\": \"processing\"}\n\n# Protect payment endpoints with combined key (user + IP)\n@app.post(\"/api/payment\")\n@throttle(\n limit=3, ttl=600, # 3 payment attempts per 10 minutes\n key_type=KeyExtractionType.COMBINED,\n key_combination=[\"header:User-ID\", \"ip\"],\n custom_error_message=\"Payment rate limit exceeded. Contact support if you need assistance.\"\n)\nasync def process_payment():\n return {\"payment_id\": \"pay_123\", \"status\": \"success\"}\n```\n\n### \ud83d\udd10 Authentication & Security\n```python\n# Login rate limiting with exponential backoff\n@app.post(\"/auth/login\")\n@throttle(\n limit=5, ttl=900, # 5 login attempts per 15 minutes\n key_type=KeyExtractionType.COMBINED,\n key_combination=[\"ip\", \"form_field:username\"],\n custom_error_message=\"Too many login attempts. Account temporarily locked.\"\n)\nasync def login():\n return {\"token\": \"jwt_token\", \"expires_in\": 3600}\n\n# Password reset protection \n@app.post(\"/auth/password-reset\")\n@throttle(\n limit=3, ttl=3600, # 3 password resets per hour\n key_type=KeyExtractionType.FORM_FIELD,\n key_field=\"email\",\n custom_error_message=\"Password reset limit exceeded. Try again in an hour.\"\n)\nasync def password_reset():\n return {\"message\": \"Password reset email sent\"}\n\n# 2FA verification\n@app.post(\"/auth/verify-2fa\")\n@throttle(\n limit=10, ttl=300, # 10 attempts per 5 minutes\n key_type=KeyExtractionType.HEADER,\n key_field=\"Session-ID\",\n custom_error_message=\"Too many 2FA verification attempts.\"\n)\nasync def verify_2fa():\n return {\"verified\": True}\n```\n\n### \ud83d\udcf1 Mobile API with Device Limits\n```python\ndef extract_device_key(request: Request) -> str:\n \"\"\"Rate limit by device fingerprint.\"\"\"\n device_id = request.headers.get(\"Device-ID\")\n app_version = request.headers.get(\"App-Version\", \"unknown\")\n platform = request.headers.get(\"Platform\", \"unknown\")\n \n if device_id:\n return f\"device:{device_id}:{platform}:{app_version}\"\n else:\n return f\"ip:{request.client.host}\"\n\n@app.get(\"/api/mobile/sync\")\n@throttle(\n limit=100, ttl=3600, # 100 syncs per hour per device\n key_type=KeyExtractionType.CUSTOM,\n key_extractor=extract_device_key\n)\nasync def mobile_sync():\n return {\"sync_data\": \"device-specific data\"}\n```\n\n### \ud83e\udd16 Bot Protection & Scraping Prevention\n```python\ndef detect_bot(request: Request) -> bool:\n \"\"\"Detect and allow verified bots.\"\"\"\n user_agent = request.headers.get(\"User-Agent\", \"\").lower()\n bot_token = request.headers.get(\"Bot-Token\")\n \n # Allow verified search engine bots\n verified_bots = [\"googlebot\", \"bingbot\", \"slurp\"]\n if any(bot in user_agent for bot in verified_bots):\n return True\n \n # Allow bots with valid tokens\n return bot_token in os.getenv(\"VALID_BOT_TOKENS\", \"\").split(\",\")\n\n@app.get(\"/api/public-data\")\n@throttle(\n limit=10, ttl=60, # Strict limits for non-bots\n bypass_function=detect_bot,\n key_type=KeyExtractionType.COMBINED,\n key_combination=[\"ip\", \"header:User-Agent\"]\n)\nasync def public_data():\n return {\"data\": \"public information\"}\n```\n\n## \ud83d\udd27 Configuration Examples\n\n### \ud83c\udf0d Environment-Based Configuration\n```python\nimport os\nfrom fastrict import RateLimitStrategy, RateLimitStrategyName\n\ndef get_rate_limit_config():\n \"\"\"Get rate limit configuration based on environment.\"\"\"\n env = os.getenv(\"ENVIRONMENT\", \"development\")\n \n if env == \"production\":\n return {\n \"strategies\": [\n RateLimitStrategy(name=RateLimitStrategyName.SHORT, limit=5, ttl=60),\n RateLimitStrategy(name=RateLimitStrategyName.MEDIUM, limit=50, ttl=600),\n RateLimitStrategy(name=RateLimitStrategyName.LONG, limit=500, ttl=3600),\n ],\n \"redis_url\": os.getenv(\"REDIS_URL\"),\n \"key_prefix\": \"prod_limits\"\n }\n elif env == \"staging\":\n return {\n \"strategies\": [\n RateLimitStrategy(name=RateLimitStrategyName.SHORT, limit=10, ttl=60),\n RateLimitStrategy(name=RateLimitStrategyName.MEDIUM, limit=100, ttl=600),\n RateLimitStrategy(name=RateLimitStrategyName.LONG, limit=1000, ttl=3600),\n ],\n \"redis_url\": os.getenv(\"REDIS_URL\", \"redis://localhost:6379/1\"),\n \"key_prefix\": \"staging_limits\"\n }\n else: # development\n return {\n \"strategies\": [\n RateLimitStrategy(name=RateLimitStrategyName.SHORT, limit=100, ttl=60),\n RateLimitStrategy(name=RateLimitStrategyName.MEDIUM, limit=1000, ttl=600),\n RateLimitStrategy(name=RateLimitStrategyName.LONG, limit=10000, ttl=3600),\n ],\n \"redis_url\": \"redis://localhost:6379/0\",\n \"key_prefix\": \"dev_limits\"\n }\n\n# Apply configuration\nconfig = get_rate_limit_config()\nrepository = RedisRateLimitRepository.from_url(\n redis_url=config[\"redis_url\"],\n key_prefix=config[\"key_prefix\"]\n)\n\napp.add_middleware(\n RateLimitMiddleware,\n rate_limit_use_case=rate_limiter,\n default_strategies=config[\"strategies\"],\n default_strategy_name=RateLimitStrategyName.MEDIUM\n)\n```\n\n### \ud83d\udccb Feature Flags Integration\n```python\ndef feature_flag_bypass(request: Request) -> bool:\n \"\"\"Bypass rate limiting based on feature flags.\"\"\"\n # Integration with feature flag service\n user_id = request.headers.get(\"User-ID\")\n \n if user_id:\n # Check if user has rate limiting bypass feature enabled\n return feature_flag_service.is_enabled(\n flag=\"rate_limiting_bypass\", \n user_id=user_id\n )\n return False\n\n@app.get(\"/api/experimental\")\n@throttle(\n limit=10, ttl=300,\n bypass_function=feature_flag_bypass\n)\nasync def experimental_feature():\n return {\"feature\": \"experimental\"}\n```\n\n## \ud83e\udd1d Contributing\n\nWe welcome contributions! Fastrict is built with \u2764\ufe0f by the community.\n\n### \ud83d\ude80 Quick Start for Contributors\n\n```bash\n# Fork and clone the repository\ngit clone https://github.com/yourusername/fastrict.git\ncd fastrict\n\n# Install development dependencies\npip install -e \".[dev]\"\n\n# Run tests\npytest\n\n# Run linting\nblack src tests\nflake8 src tests\nmypy src\n\n# Run the example\npython src/examples/simple_example.py\n```\n\n### \ud83d\udccb Contribution Guidelines\n\n- **\ud83d\udc1b Bug Reports**: Use the issue tracker with detailed reproduction steps\n- **\u2728 Feature Requests**: Propose new features with use cases\n- **\ud83d\udcdd Documentation**: Help improve our docs and examples\n- **\ud83e\uddea Tests**: Maintain 100% test coverage\n- **\ud83c\udfa8 Code Style**: Follow Ruff formatting and type hints\n\n### \ud83c\udfd7\ufe0f Development Workflow\n\n1. **Fork** the repository\n2. **Create** a feature branch: `git checkout -b feature/amazing-feature`\n3. **Make** your changes with tests\n4. **Run** the test suite: `pytest --cov=fastrict`\n5. **Commit** with clear messages: `git commit -m 'Add amazing feature'`\n6. **Push** to your fork: `git push origin feature/amazing-feature`\n7. **Create** a Pull Request\n\n## \ud83d\udcda Resources & Documentation\n\n### \ud83d\udcd6 Documentation\n- **[API Reference](https://github.com/msameim181/fastrict)** - Complete API documentation\n- **[User Guide](https://github.com/msameim181/fastrict)** - Step-by-step tutorials\n- **[Examples](https://github.com/msameim181/fastrict/tree/main/examples)** - Real-world examples\n- **[Architecture](https://github.com/msameim181/fastrict)** - Design decisions\n\n### \ud83c\udd98 Support Channels\n- **\ud83d\udc1b [Issue Tracker](https://github.com/msameim181/fastrict/issues)** - Bug reports & feature requests\n- **\ud83d\udcac [Discussions](https://github.com/msameim181/fastrict/discussions)** - Community Q&A\n- **\ud83d\udce7 [Email](mailto:9259samei@gmail.com)** - Direct support for enterprise users\n- **\ud83d\udcbc [LinkedIn](https://linkedin.com/in/msameim181)** - Professional inquiries\n\n### \ud83d\udd17 Related Projects\n- **[FastAPI](https://fastapi.tiangolo.com/)** - Modern, fast web framework for building APIs\n- **[Redis](https://redis.io/)** - In-memory data structure store \n- **[Starlette](https://www.starlette.io/)** - Lightweight ASGI framework\n- **[Pydantic](https://pydantic-docs.helpmanual.io/)** - Data validation using Python type hints\n\n## \ud83d\udcc4 License\n\nThis project is licensed under the **MIT License** - see the [LICENSE](LICENSE) file for details.\n\n## \ud83d\udcc8 Changelog & Roadmap\n\n### \ud83c\udfaf Current Version: `v0.1.1`\nSee [CHANGELOG.md](CHANGELOG.md) for version history and release notes.\n\n### \ud83d\ude80 Upcoming Features\n- **\ud83c\udf10 GraphQL Support**: Rate limiting for GraphQL endpoints\n- **\ud83c\udf10 Django Support**: Rate limiting for Django applications\n- **\ud83d\udcca Prometheus Metrics**: Built-in metrics collection\n- **\ud83d\udd04 Circuit Breaker**: Integrate with circuit breaker patterns\n- **\ud83c\udfaf Rate Limit Warming**: Gradual limit increases\n- **\ud83d\udcf1 WebSocket Support**: Rate limiting for WebSocket connections\n\n---\n\n<div align=\"center\">\n\n**Fastrict - Powering the next generation of FastAPI applications**\n\n[\u2b06\ufe0f Back to Top](#-fastrict---enterprise-fastapi-rate-limiter)\n\n</div>\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A comprehensive rate limiting system for FastAPI with Redis backend",
"version": "0.1.2",
"project_urls": {
"Bug Reports": "https://github.com/msameim181/fastrict/issues",
"Documentation": "https://fastrict.readthedocs.io/",
"Homepage": "https://github.com/msameim181/fastrict",
"Source": "https://github.com/msameim181/fastrict"
},
"split_keywords": [
"fastapi",
" rate limiting",
" redis",
" middleware",
" decorator",
" throttle"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "cd39231424607c95d76ae60c790ea0903887d867febb505c340532b61fd3060c",
"md5": "af12361f7f26c788c2d0657e9b5ec273",
"sha256": "d34d4bb599cd589d167531bbe93cdf047be2b734f0516bd564af84fb5e84d762"
},
"downloads": -1,
"filename": "fastrict-0.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "af12361f7f26c788c2d0657e9b5ec273",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 44311,
"upload_time": "2025-10-06T08:22:44",
"upload_time_iso_8601": "2025-10-06T08:22:44.326492Z",
"url": "https://files.pythonhosted.org/packages/cd/39/231424607c95d76ae60c790ea0903887d867febb505c340532b61fd3060c/fastrict-0.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "d34f53a9d2bcb5c9a36dc275251ca11cb012e4a0621109a9ff0bd585e1f285da",
"md5": "4697257b47a7e93ab6c0a3afa3f9192c",
"sha256": "100891ba2292999b2c22926da282902ead729b4b7479e133ac0eae8cb39072e5"
},
"downloads": -1,
"filename": "fastrict-0.1.2.tar.gz",
"has_sig": false,
"md5_digest": "4697257b47a7e93ab6c0a3afa3f9192c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 76104,
"upload_time": "2025-10-06T08:22:45",
"upload_time_iso_8601": "2025-10-06T08:22:45.740797Z",
"url": "https://files.pythonhosted.org/packages/d3/4f/53a9d2bcb5c9a36dc275251ca11cb012e4a0621109a9ff0bd585e1f285da/fastrict-0.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-06 08:22:45",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "msameim181",
"github_project": "fastrict",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [
{
"name": "fastapi",
"specs": [
[
">=",
"0.68.0"
]
]
},
{
"name": "redis",
"specs": [
[
">=",
"4.0.0"
]
]
},
{
"name": "pydantic",
"specs": [
[
">=",
"1.8.0"
]
]
},
{
"name": "starlette",
"specs": [
[
">=",
"0.14.0"
]
]
},
{
"name": "chromatrace",
"specs": [
[
"==",
"0.2.12"
]
]
}
],
"lcname": "fastrict"
}
JSON
