ratelimitex 0.1.0

Enhanced rate limiter with multiple strategies and monitoring capabilities

Rate Limiter

An enhanced rate limiter with a simple interface but powerful capabilities under the hood.

Features

• Decorator-based rate limiting
• Simple, intuitive API
• Sensible defaults (100 requests per minute)
• Built-in automatic error handling and retries
• Project-wide configuration
• Per-call overrides
• Multiple rate limiting strategies
• Burst handling
• Statistics and monitoring
• Dynamic adaptation based on API response headers

Rate Limiting Strategies

Strict Mode (Default)

• Never exceeds the rate limit
• Waits for the full time window when limit is reached
• Best for APIs with strict rate limits

Burst Mode

• Allows short bursts above the limit
• Configurable burst size and window
• Optional cooldown period after bursts
• Best for APIs that allow occasional bursts

Adaptive Mode

• Dynamically adjusts based on API response headers
• Adapts to changing rate limits during runtime
• Handles rate limit errors gracefully
• Best for APIs with variable rate limits or multiple environments

Installation

pip install rate-limiter

Usage

Decorator Approach (Recommended)

The simplest way to use the rate limiter is with decorators:

from rate_limiter import rate_limited

# Apply rate limiting to any async function
@rate_limited(max_requests=100, time_window=60)
async def fetch_user(user_id: str):
    return await api_client.get(f"/users/{user_id}")

# Call the function normally - rate limiting is applied automatically
user = await fetch_user("12345")

Specialized decorators for different strategies:

from rate_limiter import adaptive_rate_limited, burst_rate_limited

# Adaptive rate limiting (adjusts based on API responses)
@adaptive_rate_limited()
async def fetch_trending():
    return await api_client.get("/trending")

# Burst rate limiting (allows short bursts of higher traffic)
@burst_rate_limited(max_requests=100, burst_size=150, burst_window=10)
async def fetch_comments(post_id: str):
    return await api_client.get(f"/posts/{post_id}/comments")

Access statistics from the decorated function:

# Get rate limit statistics directly from the decorated function
stats = fetch_user.get_stats()
print(f"Total requests: {stats.total_requests}")

Client-based Approach

You can also use the explicit client approach:

from rate_limiter import RateLimitedClient

# Create a client with default settings (100 req/min)
client = RateLimitedClient()

# Use the execute method - handles all rate limiting automatically
response = await client.execute(api_call)

# Can also call methods with arguments
response = await client.execute(api_client.get, "/users", headers={"Accept": "application/json"})

Context Manager Usage

For more control, you can use the context manager:

from rate_limiter import RateLimitedClient

client = RateLimitedClient()

# Use with context manager
async with client:
    response = await api_call()

execute() vs Context Manager

The execute() method and context manager approach (async with) provide different levels of functionality:

# execute() approach - RECOMMENDED
response = await client.execute(api_call)

# Context manager approach
async with client:
    response = await api_call()

Key differences:

1. Error handling:

   • execute(): Provides robust error handling with automatic retries for rate limit errors (up to 3 retries by default)
   • Context manager: Only updates the rate limiter when errors occur but doesn't retry automatically

2. Rate limit detection:

   • execute(): Automatically detects rate limit errors through status codes (429) and error messages
   • Context manager: Captures any exception and updates the rate limiter, but doesn't distinguish rate limit errors

3. Ease of use:

   • execute(): Handles the entire request flow in one method call
   • Context manager: Requires additional error handling code from the user

For most use cases, the execute() method is recommended as it provides more comprehensive rate limit handling.

Data Models

The rate limiter uses Pydantic models for both configuration and statistics, providing better validation, documentation, and developer experience.

Configuration

from rate_limiter import RateLimitConfig, RateLimitStrategy

# Create a configuration with validation
config = RateLimitConfig(
    max_requests=100,
    time_window=60,
    strategy=RateLimitStrategy.ADAPTIVE,
    burst_size=150  # Only used with BURST strategy
)

# Access with proper type hints
print(f"Strategy: {config.strategy}")
print(f"Requests per minute: {config.max_requests / (config.time_window/60)}")

The RateLimitConfig model validates inputs (e.g., ensuring positive values for requests and time windows) and provides descriptive field documentation visible in modern IDEs.

Statistics

As shown in the Monitoring section, the get_stats() method returns a RateLimiterStats Pydantic model that provides type hints, validation, and descriptive field documentation.

Project-wide Configuration

from rate_limiter import configure

# Set project defaults
configure(
    max_requests=200,    # 200 requests
    time_window=60       # per minute
)

Per-call Overrides

from rate_limiter import RateLimitedClient

client = RateLimitedClient()

# With execute
response = await client.with_options(max_requests=50).execute(api_call)

# Or with context manager
async with client.with_options(max_requests=50):
    response = await api_call()

Adaptive Rate Limiting

The ADAPTIVE strategy can automatically adjust based on API response headers:

from rate_limiter import adaptive_rate_limited

# Use the adaptive decorator
@adaptive_rate_limited()
async def fetch_data():
    return await api_client.get("/data")

The rate limiter will automatically detect and adapt to:

• Retry-After headers
• X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset headers
• Various vendor-specific rate limit headers (GitHub, Twitter, AWS, etc.)
• Rate limit information in error messages

Custom Header Extraction

You can provide a custom function to extract headers from responses:

def extract_my_headers(response):
    """Extract headers from a custom response object"""
    headers = {}
    if hasattr(response, 'custom_headers'):
        headers = response.custom_headers
    return headers

# Use with decorator
@rate_limited(
    strategy=RateLimitStrategy.ADAPTIVE,
    extract_headers_callback=extract_my_headers
)
async def fetch_data():
    return await api_client.get("/data")

Monitoring

# Get current statistics from a decorated function
stats = fetch_data.get_stats()
print(f"Total requests: {stats.total_requests}")
print(f"Current rate: {stats.current_rate} requests/minute")
print(f"Total wait time: {stats.total_wait_time} seconds")
print(f"Rate limit hits: {stats.rate_limit_hits}")

# Check dynamic adaptations (if any)
if stats.dynamic_adjustments:
    print(f"Last update: {stats.last_dynamic_update}")
    for key, value in stats.dynamic_adjustments.items():
        print(f"  {key}: {value}")

The get_stats() method returns a RateLimiterStats Pydantic model with the following fields:

from rate_limiter import RateLimiterStats

# Required fields (always present)
stats.total_requests      # Total number of requests processed
stats.total_wait_time     # Total time spent waiting in seconds
stats.max_wait_time       # Maximum wait time encountered in seconds
stats.current_rate        # Current request rate in requests per minute
stats.current_queue_size  # Current number of requests in the tracking queue
stats.rate_limit_hits     # Number of rate limit errors encountered

# Optional fields (only present under certain conditions)
stats.last_dynamic_update       # Timestamp of last dynamic adaptation
stats.dynamic_adjustments       # Dictionary of dynamic rate limit adjustments
stats.last_rate_limit_hit       # Timestamp of last rate limit hit
stats.time_since_last_rate_limit  # Seconds since last rate limit hit
stats.rate_limit_expiry_in      # Seconds until rate limit tracking expires

Using the Pydantic model provides better type hints, validation, and IntelliSense support in modern IDEs.

Utility Functions

Detecting Rate Limit Errors

You can identify rate limit errors in your own error handling code:

from rate_limiter import is_rate_limit_error

try:
    response = await api_client.get("/endpoint")
except Exception as e:
    if is_rate_limit_error(e):
        print("Hit a rate limit, backing off...")
        await asyncio.sleep(30)
        # Retry logic
    else:
        # Handle other types of errors
        raise

The is_rate_limit_error function detects rate limit errors by checking:

• HTTP 429 status codes
• Common rate limit phrases in error messages
• Standard response patterns across different APIs

This utility can be helpful when building custom rate limit handling or for diagnostic purposes.

Contributing

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

License

This project is licensed under the MIT License - see the LICENSE file for details.