tessl/pypi-limits

Rate limiting utilities for Python with multiple strategies and storage backends

—

Pending

Overview

Eval results

Files

Rate Limiting Strategies

Name: tessl/pypi-limits
Author: tessl

Rate limiting strategies implement different algorithms for enforcing rate limits, each with distinct characteristics for accuracy, memory usage, and computational efficiency. The choice of strategy depends on your specific requirements for precision, resource usage, and acceptable trade-offs.

Capabilities

Base Rate Limiter

Abstract base class defining the common interface for all rate limiting strategies.

from abc import ABC, abstractmethod
from limits.storage import Storage
from limits.limits import RateLimitItem
from limits.util import WindowStats

class RateLimiter(ABC):
    """Abstract base class for rate limiting strategies"""
    
    def __init__(self, storage: Storage):
        """
        Initialize rate limiter with storage backend.
        
        Args:
            storage: Storage backend for persisting rate limit data
        """
    
    @abstractmethod
    def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
        """
        Consume the rate limit and return whether request is allowed.
        
        Args:
            item: The rate limit item defining limits
            identifiers: Unique identifiers for this limit instance  
            cost: Cost of this request (default: 1)
            
        Returns:
            True if request is allowed, False if rate limit exceeded
        """
    
    @abstractmethod
    def test(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
        """
        Check if rate limit allows request without consuming it.
        
        Args:
            item: The rate limit item defining limits
            identifiers: Unique identifiers for this limit instance
            cost: Expected cost to be consumed (default: 1)
            
        Returns:
            True if request would be allowed, False otherwise
        """
    
    @abstractmethod
    def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats:
        """
        Get current window statistics.
        
        Args:
            item: The rate limit item defining limits
            identifiers: Unique identifiers for this limit instance
            
        Returns:
            WindowStats with reset_time and remaining quota
        """
    
    def clear(self, item: RateLimitItem, *identifiers: str) -> None:
        """
        Clear rate limit data for the given identifiers.
        
        Args:
            item: The rate limit item defining limits
            identifiers: Unique identifiers for this limit instance
        """

Fixed Window Strategy

Divides time into fixed windows and tracks request counts per window. Simple and memory-efficient but allows bursts at window boundaries.

class FixedWindowRateLimiter(RateLimiter):
    """
    Fixed window rate limiting strategy.
    
    Divides time into fixed windows (e.g., each minute from :00 to :59).
    Allows up to the limit within each window. Memory efficient but can
    allow bursts of up to 2x the limit at window boundaries.
    """
    
    def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
        """
        Consume from the rate limit.
        
        Increments the counter for the current fixed window. If the
        increment would exceed the limit, the request is rejected.
        
        Args:
            item: Rate limit configuration
            identifiers: Unique identifiers (user ID, IP, etc.)
            cost: Cost of this request (default: 1)
            
        Returns:
            True if request allowed, False if limit exceeded
        """
    
    def test(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
        """
        Check if request would be allowed without consuming.
        
        Args:
            item: Rate limit configuration  
            identifiers: Unique identifiers
            cost: Expected cost (default: 1)
            
        Returns:
            True if request would be allowed
        """
    
    def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats:
        """
        Get current window statistics.
        
        Returns:
            WindowStats(reset_time, remaining) where reset_time is when
            the current fixed window expires
        """

Moving Window Strategy

Maintains a sliding window of individual request timestamps, providing the most accurate rate limiting but with higher memory usage.

class MovingWindowRateLimiter(RateLimiter):
    """
    Moving window rate limiting strategy.
    
    Maintains individual timestamps for each request within the time window.
    Provides the most accurate rate limiting by checking requests within
    exactly the specified time period, but uses more memory.
    
    Requires storage backend with MovingWindowSupport.
    """
    
    def __init__(self, storage: Storage):
        """
        Initialize moving window rate limiter.
        
        Args:
            storage: Storage backend that supports acquire_entry and 
                    get_moving_window methods
                    
        Raises:
            NotImplementedError: If storage doesn't support moving window
        """
    
    def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
        """
        Consume from the rate limit using moving window.
        
        Adds timestamp entries for the request and removes expired ones.
        Checks if total requests in the sliding window exceed the limit.
        
        Args:
            item: Rate limit configuration
            identifiers: Unique identifiers
            cost: Cost of this request (default: 1)
            
        Returns:
            True if request allowed, False if limit exceeded
        """
    
    def test(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
        """
        Check if request would be allowed in moving window.
        
        Args:
            item: Rate limit configuration
            identifiers: Unique identifiers  
            cost: Expected cost (default: 1)
            
        Returns:
            True if request would be allowed
        """
    
    def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats:
        """
        Get moving window statistics.
        
        Returns:
            WindowStats(reset_time, remaining) where reset_time is when
            the oldest request in the window expires
        """

Sliding Window Counter Strategy

Approximates a moving window using weighted counters from current and previous fixed windows. Balances accuracy and memory efficiency.

class SlidingWindowCounterRateLimiter(RateLimiter):
    """
    Sliding window counter rate limiting strategy.
    
    Approximates a moving window by weighting the current and previous
    fixed window counters based on time elapsed. More memory efficient
    than moving window while more accurate than fixed window.
    
    Added in version 4.1. Requires storage with SlidingWindowCounterSupport.
    """
    
    def __init__(self, storage: Storage):
        """
        Initialize sliding window counter rate limiter.
        
        Args:
            storage: Storage backend supporting get_sliding_window and
                    acquire_sliding_window_entry methods
                    
        Raises:
            NotImplementedError: If storage doesn't support sliding window counter
        """
    
    def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
        """
        Consume from rate limit using sliding window counter.
        
        Uses weighted combination of current and previous window counters
        to approximate the moving window behavior.
        
        Args:
            item: Rate limit configuration
            identifiers: Unique identifiers
            cost: Cost of this request (default: 1)
            
        Returns:
            True if request allowed, False if limit exceeded
        """
    
    def test(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
        """
        Check if request would be allowed using sliding window counter.
        
        Args:
            item: Rate limit configuration
            identifiers: Unique identifiers
            cost: Expected cost (default: 1)
            
        Returns:
            True if request would be allowed
        """
    
    def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats:
        """
        Get sliding window counter statistics.
        
        Returns:
            WindowStats(reset_time, remaining) with approximated values
            based on weighted counters
        """

Legacy Fixed Window with Elastic Expiry (Deprecated)

Fixed window strategy with elastic expiry behavior. Deprecated in version 4.1.

class FixedWindowElasticExpiryRateLimiter(FixedWindowRateLimiter):
    """
    Fixed window with elastic expiry rate limiter.
    
    Similar to FixedWindowRateLimiter but with elastic expiry behavior.
    
    Deprecated in version 4.1 - use FixedWindowRateLimiter instead.
    """
    
    def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
        """
        Consume from rate limit with elastic expiry.
        
        Args:
            item: Rate limit configuration
            identifiers: Unique identifiers
            cost: Cost of this request (default: 1)
            
        Returns:
            True if request allowed, False if limit exceeded
        """

Strategy Selection

# Type alias for all known strategy types
KnownStrategy = (
    type[SlidingWindowCounterRateLimiter]
    | type[FixedWindowRateLimiter]
    | type[FixedWindowElasticExpiryRateLimiter]
    | type[MovingWindowRateLimiter]
)

# Strategy registry mapping names to classes
STRATEGIES: dict[str, KnownStrategy] = {
    "fixed-window": FixedWindowRateLimiter,
    "moving-window": MovingWindowRateLimiter, 
    "sliding-window-counter": SlidingWindowCounterRateLimiter,
    "fixed-window-elastic-expiry": FixedWindowElasticExpiryRateLimiter
}

Usage Examples

Basic Strategy Usage

from limits import RateLimitItemPerMinute
from limits.storage import storage_from_string
from limits.strategies import (
    FixedWindowRateLimiter,
    MovingWindowRateLimiter,
    SlidingWindowCounterRateLimiter
)

# Set up rate limit and storage
rate_limit = RateLimitItemPerMinute(100)  # 100 requests per minute
storage = storage_from_string("redis://localhost:6379")

# Choose strategy based on requirements
fixed_limiter = FixedWindowRateLimiter(storage)         # Memory efficient
moving_limiter = MovingWindowRateLimiter(storage)       # Most accurate
sliding_limiter = SlidingWindowCounterRateLimiter(storage)  # Balanced

user_id = "user123"

# Test and consume with any strategy
if fixed_limiter.test(rate_limit, user_id):
    success = fixed_limiter.hit(rate_limit, user_id)
    if success:
        print("Request allowed")
    else:
        print("Rate limit exceeded")

# Get window statistics
stats = fixed_limiter.get_window_stats(rate_limit, user_id)
print(f"Remaining: {stats.remaining}")
print(f"Reset time: {stats.reset_time}")

Strategy Comparison

import time
from limits import RateLimitItemPerMinute
from limits.storage import MemoryStorage
from limits.strategies import (
    FixedWindowRateLimiter,
    MovingWindowRateLimiter, 
    SlidingWindowCounterRateLimiter
)

# Create rate limit and storage
rate_limit = RateLimitItemPerMinute(10)  # 10 requests per minute
storage = MemoryStorage()

# Initialize all strategies
fixed = FixedWindowRateLimiter(storage)
moving = MovingWindowRateLimiter(storage) 
sliding = SlidingWindowCounterRateLimiter(storage)

user_id = "test_user"

# Fixed Window: May allow bursts at window boundaries
for i in range(15):
    result = fixed.hit(rate_limit, f"{user_id}_fixed", i)
    print(f"Fixed window request {i}: {'allowed' if result else 'denied'}")

# Moving Window: Precise tracking of each request timestamp  
for i in range(15):
    result = moving.hit(rate_limit, f"{user_id}_moving", i)
    print(f"Moving window request {i}: {'allowed' if result else 'denied'}")
    time.sleep(0.1)  # Small delay between requests

# Sliding Window Counter: Approximation using weighted counters
for i in range(15):
    result = sliding.hit(rate_limit, f"{user_id}_sliding", i)
    print(f"Sliding window request {i}: {'allowed' if result else 'denied'}")

Working with Costs

from limits import RateLimitItemPerSecond
from limits.storage import MemoryStorage
from limits.strategies import FixedWindowRateLimiter

# Rate limit allowing 100 "cost units" per second
rate_limit = RateLimitItemPerSecond(100)
storage = MemoryStorage()
limiter = FixedWindowRateLimiter(storage)

api_key = "api_key_123"

# Different operations have different costs
light_request_cost = 1    # GET requests
heavy_request_cost = 10   # POST requests with file upload
batch_request_cost = 50   # Batch operations

# Check if expensive operation is allowed
if limiter.test(rate_limit, api_key, cost=heavy_request_cost):
    success = limiter.hit(rate_limit, api_key, cost=heavy_request_cost)
    if success:
        print("Heavy request processed")

# Check remaining quota
stats = limiter.get_window_stats(rate_limit, api_key)
print(f"Remaining cost units: {stats.remaining}")

Install with Tessl CLI