throtty 0.0.1

Rate limit library for ASGI application

Throtty

A flexible and powerful rate limiting library for ASGI applications like FastAPI and Starlette.

Features

• 🚀 Easy Integration - Simple decorator-based API for quick setup
• 🔄 Multiple Algorithms - Support for sliding window counter, sliding window log, and token bucket
• 💾 Flexible Storage - In-memory storage or Redis for distributed systems
• 🎯 Pattern Matching - Regex and wildcard support for flexible path matching
• 🔑 Custom Key Extraction - Rate limit by IP, user ID, API key, or any custom logic
• 🌐 Distributed Ready - Redis backend for multi-server deployments
• 📊 Informative Headers - Standard rate limit headers in responses

Installation

pip install throtty

Quick Start

from fastapi import FastAPI
from throtty.core import Throtty, rule

# Initialize Throtty with in-memory storage
limiter = Throtty()

app = FastAPI()
limiter.install(app)

# Apply rate limit: 10 requests per minute
@app.get("/api/limited")
@rule("/api/limited", "10/60")
async def limited_endpoint():
    return {"message": "This endpoint is rate limited"}

Basic Usage

1. Initialize Throtty

In-Memory Storage (Single Server)

from throtty.core import Throtty

limiter = Throtty()

Redis Storage (Distributed/Multi-Server)

# Using Redis DSN
limiter = Throtty(redis_dsn="redis://localhost:6379/0")

# Using existing Redis client
from redis import Redis
redis_client = Redis(host='localhost', port=6379, db=0)
limiter = Throtty(redis=redis_client)

# Using Redis connection pool
from redis import ConnectionPool
pool = ConnectionPool(host='localhost', port=6379, db=0)
limiter = Throtty(redis_pool=pool)

2. Install Middleware

from fastapi import FastAPI

app = FastAPI()
limiter.install(app)  # Must be called to activate rate limiting

3. Add Rate Limiting Rules

Using Decorators (Recommended)

from throtty.core import rule

@app.get("/api/users")
@rule("/api/users", "100/60")  # 100 requests per minute
async def get_users():
    return {"users": []}

Using add_rule Method

# Add rules programmatically
limiter.add_rule("/api/posts", limit=50, window=60)
limiter.add_rule("/api/comments", limit=200, window=3600)

Advanced Usage

Multiple Rate Limits

Apply multiple rate limits to the same endpoint (all must pass):

@app.post("/api/data")
@rule("/api/data", "10/60;100/3600")  # 10/min AND 100/hour
async def post_data():
    return {"status": "created"}

Path Patterns

Exact Match

@rule("/api/users", "100/60")  # Only matches /api/users

Wildcard Match

@rule("/api/v1/*", "1000/3600")  # Matches /api/v1/users, /api/v1/posts, etc.

Regex Pattern

@rule("^/api/v[0-9]+/.*", "2000/3600")  # Matches /api/v1/..., /api/v2/..., etc.

Custom Key Extraction

Rate limit by user ID instead of IP address:

def extract_user_id(host, headers):
    """Extract user ID from authorization header"""
    user_id = headers.get('x-user-id', 'anonymous')
    return f"user:{user_id}"

@app.get("/api/profile")
@rule("/api/profile", "50/3600", key_func=extract_user_id)
async def get_profile():
    return {"profile": "data"}

Global Key Extractor

Set a default key extractor for all rules:

def extract_api_key(host, headers):
    api_key = headers.get('x-api-key', 'anonymous')
    return f"apikey:{api_key}"

limiter.set_key_extractor(extract_api_key)

# All rules will now use API key unless overridden
@app.get("/api/data")
@rule("/api/data", "100/60")  # Uses global extractor
async def get_data():
    return {"data": []}

Rate Limiting Algorithms

Choose the algorithm that best fits your needs:

# Sliding Window Counter (default) - Best balance of accuracy and performance
limiter = Throtty(algorithm="slidingwindow_counter")

# Sliding Window Log - Most accurate, higher memory usage
limiter = Throtty(algorithm="slidingwindow_log")

# Token Bucket - Allows bursts, smooth rate limiting
limiter = Throtty(algorithm="token_bucket")

Algorithm Comparison:

Algorithm	Accuracy	Memory	Performance	Bursts
Sliding Window Counter	High	Low	Excellent	No
Sliding Window Log	Highest	High	Good	No
Token Bucket	Medium	Low	Excellent	Yes

Response Headers

When a request is rate limited (HTTP 429), Throtty includes informative headers:

HTTP/1.1 429 Too Many Requests
Content-Type: application/json
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset-At: 1699564800
Retry-After: 45

"Rate limit exceeded"

Header Descriptions:

• X-RateLimit-Limit - Maximum requests allowed in the time window
• X-RateLimit-Remaining - Requests remaining in current window
• X-RateLimit-Reset-At - Unix timestamp when the rate limit resets
• Retry-After - Seconds to wait before retrying

Complete Example

from fastapi import FastAPI, Header
from throtty.core import Throtty, rule
from typing import Optional

# Initialize with Redis for distributed systems
limiter = Throtty(
    redis_dsn="redis://localhost:6379/0",
    algorithm="slidingwindow_counter"
)

app = FastAPI()
limiter.install(app)

# Custom key extractor for user-based limiting
def extract_user(host, headers):
    user_id = headers.get('authorization')
    if user_id:
        return f"user:{user_id}"
    return f"ip:{host}"

limiter.set_key_extractor(extract_user)

# Public endpoint - strict limit
@app.get("/api/public")
@rule("/api/public", "10/60")  # 10 requests per minute
async def public_endpoint():
    return {"message": "Public data"}

# Authenticated endpoint - higher limit
@app.get("/api/users")
@rule("/api/users", "100/60;1000/3600")  # 100/min AND 1000/hour
async def get_users(authorization: Optional[str] = Header(None)):
    return {"users": ["user1", "user2"]}

# Premium endpoint - custom key function
def premium_key(host, headers):
    tier = headers.get('x-subscription-tier', 'free')
    user = headers.get('x-user-id', 'anonymous')
    return f"{tier}:{user}"

@app.get("/api/premium")
@rule("/api/premium", "1000/3600", key_func=premium_key)
async def premium_endpoint():
    return {"data": "premium content"}

# Wildcard - all v1 endpoints share limit
@app.get("/api/v1/posts")
@app.get("/api/v1/comments")
@rule("/api/v1/*", "5000/3600")
async def v1_endpoints():
    return {"version": "v1"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Configuration Options

Throtty Initialization

Throtty(
    redis=None,                    # Redis client instance
    redis_pool=None,               # Redis connection pool
    redis_dsn=None,                # Redis DSN string
    max_connections=10,            # Max connections in pool
    algorithm="slidingwindow_counter"  # Rate limiting algorithm
)

Adding Rules

limiter.add_rule(
    path="/api/endpoint",          # Endpoint path (supports regex)
    limit=100,                     # Max requests
    window=60,                     # Time window in seconds
    key_func=None                  # Optional key extractor
)

Rule Decorator

@rule(
    path="/api/endpoint",          # Endpoint path
    str_rule="10/60;100/3600",    # "limit/window" format
    key_func=None                  # Optional key extractor
)

Common Patterns

Per-User Rate Limiting

def by_user_id(host, headers):
    return f"user:{headers.get('x-user-id', host)}"

@app.get("/api/user-data")
@rule("/api/user-data", "100/3600", key_func=by_user_id)
async def user_data():
    return {"data": []}

API Key Based Limiting

def by_api_key(host, headers):
    api_key = headers.get('x-api-key', 'anonymous')
    return f"apikey:{api_key}"

limiter.set_key_extractor(by_api_key)

Tiered Rate Limiting

def tiered_limit(host, headers):
    tier = headers.get('x-tier', 'free')
    user = headers.get('x-user-id', host)
    return f"{tier}:{user}"

@app.get("/api/tiered")
@rule("/api/tiered", "10/60", key_func=tiered_limit)  # Free tier
async def tiered_endpoint():
    return {"data": []}

Version-Specific Limits

# v1 endpoints - lower limit
limiter.add_rule("/api/v1/*", limit=100, window=3600)

# v2 endpoints - higher limit
limiter.add_rule("/api/v2/*", limit=1000, window=3600)

Best Practices

1. Initialize Early - Create Throtty instance before defining routes
2. Install Middleware - Always call limiter.install(app) to activate rate limiting
3. Use Redis for Production - In-memory storage only suitable for single-server setups
4. Match Paths Carefully - Ensure decorator paths match route paths
5. Test Limits - Verify rate limits work as expected in staging
6. Monitor Usage - Track rate limit hits to adjust limits appropriately
7. Graceful Degradation - Handle rate limit errors in client applications
8. Document Limits - Inform API users about rate limits in documentation

Error Handling

When rate limits are exceeded, Throtty returns a 429 response:

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(429)
async def rate_limit_handler(request: Request, exc):
    return JSONResponse(
        status_code=429,
        content={
            "error": "Rate limit exceeded",
            "message": "Please try again later"
        }
    )

Performance Considerations

• In-Memory Storage: Fast but not shared across servers
• Redis Storage: Slight network overhead but enables distributed limiting
• Algorithm Choice:
  • Use slidingwindow_counter for best performance
  • Use slidingwindow_log for highest accuracy
  • Use token_bucket for burst tolerance

Troubleshooting

Rate Limiting Not Working

1. Ensure limiter.install(app) is called
2. Verify path in @rule matches route path
3. Check Throtty is initialized before using @rule

Redis Connection Issues

# Test Redis connection
limiter = Throtty(redis_dsn="redis://localhost:6379/0")
# Check Redis is running: redis-cli ping

Rules Not Matching

# Debug rule matching
rule = limiter._find_match_rule("/api/users")
print(f"Matched rule: {rule}")

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

MIT License - see LICENSE file for details

Support

• Issues: GitHub Issues
• Discussions: GitHub Discussions

Changelog

v0.0.1

• Initial release
• Support for FastAPI
• In-memory and Redis storage
• Multiple rate limiting algorithms
• Flexible path matching with regex
• Custom key extraction functions