or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

core-limiting.mdfactory-patterns.mdindex.mdrate-configuration.mdstorage-backends.mdtime-sources.md
tile.json

tessl/pypi-pyrate-limiter

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

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/pyrate-limiter@3.9.x

To install, run

npx @tessl/cli install tessl/pypi-pyrate-limiter@3.9.0

index.mddocs/

PyrateLimiter

A comprehensive Python rate-limiting library implementing the Leaky Bucket algorithm for controlling request rates in applications. PyrateLimiter supports unlimited rate limits with custom intervals, separately tracks limits for different services or resources, and manages limit breaches through configurable exception raising or delay mechanisms.

The library offers both synchronous and asynchronous workflows with multiple backend options including in-memory, Redis, SQLite, and PostgreSQL storage for persistent rate limiting across threads, processes, or application restarts. Key features include decorator support for easy function wrapping, multiprocessing compatibility, weight-based item consumption, and extensible architecture allowing custom bucket implementations.

Package Information

  • Package Name: pyrate-limiter
  • Language: Python
  • Installation: pip install pyrate-limiter
  • Version: 3.9.0
  • License: MIT

Core Imports

import pyrate_limiter

Common usage imports:

from pyrate_limiter import Limiter, Rate, Duration, BucketFullException
from pyrate_limiter import InMemoryBucket, SQLiteBucket, RedisBucket

Basic Usage

from pyrate_limiter import Limiter, Rate, Duration

# Create a limiter: 5 requests per second
limiter = Limiter(Rate(5, Duration.SECOND))

# Try to acquire permission for a request
if limiter.try_acquire("user123"):
    print("Request allowed")
else:
    print("Rate limit exceeded")

# Using with context manager
with Limiter(Rate(10, Duration.MINUTE)) as limiter:
    success = limiter.try_acquire("api_call", weight=2)

# Using as decorator
@limiter.as_decorator()
def rate_limited_function(user_id):
    return (user_id, 1)  # (name, weight) for rate limiting

@rate_limited_function
def api_call(user_id):
    return f"Processing request for {user_id}"

Architecture

PyrateLimiter uses a modular architecture based on three core abstractions:

  • Rate: Defines rate limit rules (limit/interval pairs)
  • Bucket: Storage backend for tracking rate limit state (in-memory, SQLite, Redis, PostgreSQL)
  • Clock: Time source for rate calculations (system time, monotonic, remote database time)
  • Limiter: Orchestrates rate limiting logic using buckets and clocks

This design enables flexible deployment scenarios from simple in-process rate limiting to distributed systems with shared state across multiple application instances.

Capabilities

Core Rate Limiting

The main Limiter class provides the primary rate limiting functionality with support for both synchronous and asynchronous operations, configurable delays, and exception handling.

class Limiter:
    def __init__(
        self,
        argument: Union[BucketFactory, AbstractBucket, Rate, List[Rate]],
        clock: Optional[AbstractClock] = None,
        raise_when_fail: bool = True,
        max_delay: Optional[Union[int, Duration]] = None,
        retry_until_max_delay: bool = False,
        buffer_ms: int = 50
    ): ...
    
    def try_acquire(self, name: str, weight: int = 1) -> Union[bool, Awaitable[bool]]: ...
    
    async def try_acquire_async(self, name: str, weight: int = 1) -> bool: ...
    
    def as_decorator(self) -> Callable[[ItemMapping], DecoratorWrapper]: ...

Core Rate Limiting

Storage Backends

Multiple storage backends for persisting rate limit state across application restarts, processes, and distributed deployments.

class InMemoryBucket(AbstractBucket):
    def __init__(self, rates: List[Rate]): ...

class SQLiteBucket(AbstractBucket):
    def __init__(self, rates: List[Rate], conn: sqlite3.Connection, table: str, lock: Optional[RLock] = None): ...
    
    @classmethod
    def init_from_file(
        cls, rates: List[Rate], db_path: str, table: str = "pyrate_limiter",
        create_new_table: bool = False, use_file_lock: bool = False
    ) -> "SQLiteBucket": ...

class RedisBucket(AbstractBucket):
    def __init__(self, rates: List[Rate], redis: Union[Redis, AsyncRedis], bucket_key: Optional[str] = None): ...

Storage Backends

Rate Configuration

Rate definition and duration utilities for configuring rate limits with flexible time intervals and validation.

class Rate:
    def __init__(self, limit: int, interval: Union[int, Duration]): ...

class Duration(Enum):
    SECOND = 1000
    MINUTE = 60000
    HOUR = 3600000
    DAY = 86400000
    WEEK = 604800000

Rate Configuration

Factory Patterns

Pre-configured factory functions and patterns for common use cases including multiprocessing support and simplified bucket creation.

def create_limiter(
    rate_per_duration: int = 3,
    duration: Union[int, Duration] = Duration.SECOND,
    clock: Optional[AbstractClock] = None,
    **limiter_kwargs
) -> Limiter: ...

def create_sqlite_limiter(
    rate_per_duration: int = 3,
    duration: Union[int, Duration] = Duration.SECOND,
    db_path: Optional[str] = None,
    **kwargs
) -> Limiter: ...

Factory Patterns

Time Sources

Different clock implementations for various deployment scenarios including local system time, monotonic time, and remote database-backed time sources.

class TimeClock(AbstractClock):
    def now(self) -> int: ...

class MonotonicClock(AbstractClock):
    def now(self) -> int: ...

class SQLiteClock(AbstractClock):
    def __init__(self, conn: Union[sqlite3.Connection, SQLiteBucket]): ...

Time Sources

Exception Handling

class BucketFullException(Exception):
    def __init__(self, item: RateItem, rate: Rate): ...
    item: RateItem
    rate: Rate
    meta_info: Dict[str, Union[str, float]]

class LimiterDelayException(Exception):
    def __init__(self, item: RateItem, rate: Rate, actual_delay: int, max_delay: int): ...
    item: RateItem
    rate: Rate
    actual_delay: int
    max_delay: int
    meta_info: Dict[str, Union[str, float]]

Utility Functions

Utility functions for rate limiting operations and validation.

def binary_search(items: List[RateItem], value: int) -> int:
    """
    Find the index of item in list where left.timestamp < value <= right.timestamp.
    
    Used internally to determine current size of time windows for rate calculations.
    
    Parameters:
    - items: List of RateItem objects sorted by timestamp
    - value: Timestamp value to search for
    
    Returns:
    - int: Index position for the timestamp boundary
    """

def validate_rate_list(rates: List[Rate]) -> bool:
    """
    Validate that rates are correctly ordered for rate limiting.
    
    Parameters:
    - rates: List of Rate objects to validate
    
    Returns:
    - bool: True if rates are valid, False otherwise
    
    Validation rules:
    - Rates must be ordered by increasing interval
    - Rates must be ordered by increasing limit
    - Rate density (limit/interval) must be decreasing
    """

def id_generator(size: int = 6, chars: str = ...) -> str:
    """
    Generate random string identifier.
    
    Parameters:
    - size: Length of generated string (default: 6)
    - chars: Character set to use (default: alphanumeric)
    
    Returns:
    - str: Random string identifier
    """

def dedicated_sqlite_clock_connection() -> sqlite3.Connection:
    """
    Create dedicated SQLite connection for clock operations.
    
    Returns:
    - sqlite3.Connection: Configured SQLite connection for time queries
    
    Features:
    - Uses temporary database file
    - Configured for EXCLUSIVE isolation
    - Thread-safe with check_same_thread=False
    """

Core Types

class RateItem:
    def __init__(self, name: str, timestamp: int, weight: int = 1): ...
    name: str
    weight: int
    timestamp: int

ItemMapping = Callable[[Any], Tuple[str, int]]
DecoratorWrapper = Callable[[Callable[[Any], Any]], Callable[[Any], Any]]