Python Rate-Limiter using Leaky-Bucket Algorithm for controlling request rates in applications with multiple backend storage options.
81
Rate definition and duration utilities for configuring rate limits with flexible time intervals and validation. The Rate class defines the core rate limiting rules while Duration provides convenient time constants.
Core class for defining rate limit rules with limit and interval parameters.
class Rate:
def __init__(self, limit: int, interval: Union[int, Duration]):
"""
Define a rate limit rule.
Parameters:
- limit: Number of requests allowed within the interval
- interval: Time interval in milliseconds (or Duration enum)
Properties:
- limit: int - Maximum number of requests
- interval: int - Time interval in milliseconds
"""
def __str__(self) -> str:
"""Human-readable rate representation."""
def __repr__(self) -> str:
"""Developer representation of rate."""Usage example:
from pyrate_limiter import Rate, Duration
# Basic rate definitions
rate1 = Rate(10, Duration.SECOND) # 10 requests per second
rate2 = Rate(100, Duration.MINUTE) # 100 requests per minute
rate3 = Rate(1000, Duration.HOUR) # 1000 requests per hour
# Using milliseconds directly
rate4 = Rate(5, 1000) # 5 requests per 1000ms (1 second)
# Multiple rates for complex limiting
rates = [
Rate(10, Duration.SECOND), # Burst protection
Rate(100, Duration.MINUTE), # Short-term limit
Rate(1000, Duration.HOUR), # Long-term limit
]
print(str(rate1)) # "limit=10/1.0s"
print(str(rate2)) # "limit=100/1.0m"Convenient time interval constants with arithmetic operations support.
class Duration(Enum):
SECOND = 1000
MINUTE = 60000
HOUR = 3600000
DAY = 86400000
WEEK = 604800000
def __mul__(self, multiplier: float) -> int:
"""Multiply duration by a factor."""
def __rmul__(self, multiplier: float) -> int:
"""Reverse multiply duration by a factor."""
def __add__(self, another_duration: Union["Duration", int]) -> int:
"""Add duration to another duration or integer."""
def __radd__(self, another_duration: Union["Duration", int]) -> int:
"""Reverse add duration to another duration or integer."""
def __int__(self) -> int:
"""Convert duration to integer milliseconds."""
def __eq__(self, duration: object) -> bool:
"""Compare duration equality."""
@staticmethod
def readable(value: int) -> str:
"""Convert milliseconds to human-readable format."""Usage example:
from pyrate_limiter import Duration, Rate
# Basic duration usage
print(Duration.SECOND) # Duration.SECOND
print(int(Duration.SECOND)) # 1000
print(Duration.MINUTE) # Duration.MINUTE
print(int(Duration.MINUTE)) # 60000
# Arithmetic operations
half_minute = Duration.MINUTE * 0.5 # 30000ms
two_hours = Duration.HOUR * 2 # 7200000ms
day_and_hour = Duration.DAY + Duration.HOUR # 90000000ms
# Using with Rate
rate = Rate(50, Duration.MINUTE * 5) # 50 requests per 5 minutes
rate = Rate(10, Duration.SECOND * 30) # 10 requests per 30 seconds
# Human-readable formatting
print(Duration.readable(1000)) # "1.0s"
print(Duration.readable(60000)) # "1.0m"
print(Duration.readable(3600000)) # "1.0h"
print(Duration.readable(86400000)) # "1.0d"
print(Duration.readable(1500)) # "1.5s"Wrapper class for individual rate-limited items with timestamps and weights.
class RateItem:
def __init__(self, name: str, timestamp: int, weight: int = 1):
"""
Create a rate-limited item.
Parameters:
- name: Identifier for the item/resource
- timestamp: Timestamp in milliseconds
- weight: Weight/cost of the item (default: 1)
Properties:
- name: str - Item identifier
- timestamp: int - Item timestamp in milliseconds
- weight: int - Item weight/cost
"""
def __str__(self) -> str:
"""String representation of rate item."""Usage example:
from pyrate_limiter import RateItem
import time
# Create rate items
current_time = int(time.time() * 1000)
item1 = RateItem("user123", current_time)
item2 = RateItem("expensive_op", current_time, weight=5)
print(item1) # RateItem(name=user123, weight=1, timestamp=...)
print(item2) # RateItem(name=expensive_op, weight=5, timestamp=...)
# Rate items are used internally by Limiter
# but can be created manually for advanced use casesUtility function for validating rate list ordering and consistency.
def validate_rate_list(rates: List[Rate]) -> bool:
"""
Raise false if rates are incorrectly ordered.
Parameters:
- rates: List of rate configurations to validate
Returns:
- bool: True if rates are valid, False otherwise
Validation rules:
- Intervals must be in ascending order
- Limits must be in ascending order
- Rate ratios (limit/interval) must be in descending order
"""Usage example:
from pyrate_limiter import Rate, Duration, validate_rate_list
# Valid rate list (intervals and limits ascending, ratios descending)
valid_rates = [
Rate(10, Duration.SECOND), # 10/1000ms = 0.01 req/ms
Rate(100, Duration.MINUTE), # 100/60000ms = 0.00167 req/ms
Rate(1000, Duration.HOUR), # 1000/3600000ms = 0.00028 req/ms
]
# Invalid rate list
invalid_rates = [
Rate(100, Duration.MINUTE), # Wrong order
Rate(10, Duration.SECOND),
]
print(validate_rate_list(valid_rates)) # True
print(validate_rate_list(invalid_rates)) # FalseCombine short and long interval rates for burst protection with sustained limits.
from pyrate_limiter import Rate, Duration
# Allow bursts of 10/second but limit to 100/minute overall
burst_and_sustained = [
Rate(10, Duration.SECOND), # Burst protection
Rate(100, Duration.MINUTE), # Sustained rate
]Use multiplication for fractional rates.
from pyrate_limiter import Rate, Duration
# 1 request every 2 seconds (0.5 requests per second)
slow_rate = Rate(1, Duration.SECOND * 2)
# 30 requests every 30 seconds (1 request per second average)
averaged_rate = Rate(30, Duration.SECOND * 30)Different operations can have different weights/costs.
from pyrate_limiter import Limiter, Rate, Duration
limiter = Limiter(Rate(100, Duration.MINUTE))
# Light operations (weight=1, default)
limiter.try_acquire("read_operation")
# Heavy operations (weight=10)
limiter.try_acquire("expensive_computation", weight=10)
# Bulk operations (weight=50)
limiter.try_acquire("batch_processing", weight=50)Install with Tessl CLI
npx tessl i tessl/pypi-pyrate-limiterdocs
evals
scenario-1
scenario-2
scenario-3
scenario-4
scenario-5
scenario-6
scenario-7
scenario-8
scenario-9
scenario-10