CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-pyrate-limiter

Python Rate-Limiter using Leaky-Bucket Algorithm for controlling request rates in applications with multiple backend storage options.

81

1.44x
Overview
Eval results
Files

rate-configuration.mddocs/

Rate Configuration

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.

Capabilities

Rate Definition

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"

Duration Constants

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"

Rate Item

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 cases

Rate Validation

Utility 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))  # False

Advanced Rate Patterns

Burst and Sustained Rates

Combine 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
]

Fractional Rates

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)

Weighted Operations

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-limiter

docs

core-limiting.md

factory-patterns.md

index.md

rate-configuration.md

storage-backends.md

time-sources.md

tile.json