or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

async.mdindex.mdrate-limit-items.mdstorage.mdstrategies.mdutilities.md
tile.json

tessl/pypi-limits

Rate limiting utilities for Python with multiple strategies and storage backends

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/limits@4.8.x

To install, run

npx @tessl/cli install tessl/pypi-limits@4.8.0

index.mddocs/

Limits

A comprehensive Python library for implementing rate limiting functionality across various storage backends. Limits provides multiple rate limiting strategies including Fixed Window, Moving Window, and Sliding Window Counter algorithms, each with different memory efficiency and accuracy trade-offs.

Package Information

  • Package Name: limits
  • Language: Python
  • Installation: pip install limits
  • Repository: https://github.com/alisaifee/limits
  • Documentation: https://limits.readthedocs.io/

Core Imports

import limits

Common imports for rate limiting:

from limits import RateLimitItem, parse
from limits.storage import storage_from_string
from limits.strategies import FixedWindowRateLimiter

For asynchronous usage:

from limits.aio.strategies import FixedWindowRateLimiter as AsyncFixedWindowRateLimiter
from limits.storage import storage_from_string  # Same function, use "async+" prefix

Basic Usage

from limits import RateLimitItemPerMinute, parse
from limits.storage import storage_from_string
from limits.strategies import FixedWindowRateLimiter

# Create rate limit items
rate_limit = RateLimitItemPerMinute(10)  # 10 requests per minute
# Or parse from string
rate_limit = parse("10/minute")

# Set up storage backend
storage = storage_from_string("memory://")
# Or use Redis
# storage = storage_from_string("redis://localhost:6379")

# Create rate limiter with chosen strategy
limiter = FixedWindowRateLimiter(storage)

# Apply rate limiting
user_id = "user123"
if limiter.test(rate_limit, user_id):
    # Request allowed
    limiter.hit(rate_limit, user_id)
    print("Request processed")
else:
    print("Rate limit exceeded")

# Check remaining quota
stats = limiter.get_window_stats(rate_limit, user_id)
print(f"Remaining: {stats.remaining}, Reset at: {stats.reset_time}")

Architecture

The limits library follows a modular architecture with three core components:

  • Rate Limit Items: Define the rate limit parameters (amount, time window, granularity)
  • Storage Backends: Handle persistence and distribution of rate limit counters across different storage systems
  • Rate Limiting Strategies: Implement different algorithms for rate limit enforcement

This design enables flexible combinations of time windows, storage backends, and algorithmic strategies based on specific requirements for accuracy, memory efficiency, and distribution needs.

Capabilities

Rate Limit Items

Core classes for defining rate limits with different time granularities, supporting custom namespaces and time multipliers for flexible rate limiting configurations.

class RateLimitItem:
    def __init__(self, amount: int, multiples: int | None = 1, namespace: str = "LIMITER"): ...
    def get_expiry(self) -> int: ...
    def key_for(self, *identifiers: str | int | float | bytes) -> str: ...

class RateLimitItemPerSecond(RateLimitItem): ...
class RateLimitItemPerMinute(RateLimitItem): ...
class RateLimitItemPerHour(RateLimitItem): ...
class RateLimitItemPerDay(RateLimitItem): ...
class RateLimitItemPerMonth(RateLimitItem): ...
class RateLimitItemPerYear(RateLimitItem): ...

Rate Limit Items

Rate Limiting Strategies

Different algorithmic approaches for enforcing rate limits, each with distinct characteristics for accuracy, memory usage, and computational efficiency.

class FixedWindowRateLimiter:
    def __init__(self, storage: Storage): ...
    def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool: ...
    def test(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool: ...
    def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats: ...

class MovingWindowRateLimiter:
    def __init__(self, storage: Storage): ...
    def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool: ...
    def test(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool: ...
    def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats: ...

class SlidingWindowCounterRateLimiter:
    def __init__(self, storage: Storage): ...
    def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool: ...
    def test(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool: ...
    def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats: ...

Rate Limiting Strategies

Storage Backends

Multiple storage implementations supporting both local and distributed rate limiting with various backends including in-memory, Redis, Memcached, MongoDB, and etcd.

def storage_from_string(storage_string: str, **options: float | str | bool) -> Storage: ...

class MemoryStorage(Storage): ...
class RedisStorage(Storage): ...
class RedisClusterStorage(Storage): ...
class RedisSentinelStorage(Storage): ...
class MemcachedStorage(Storage): ...
class MongoDBStorage(Storage): ...
class EtcdStorage(Storage): ...

Storage Backends

Asynchronous Support

Complete async/await API providing the same functionality as the synchronous API but optimized for asynchronous applications and frameworks.

# Async strategies mirror sync API but with async methods
class FixedWindowRateLimiter:
    def __init__(self, storage: Storage): ...
    async def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool: ...
    async def test(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool: ...
    async def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats: ...

# Async storage backends with same interface as sync versions
class RedisStorage(Storage): ...
class MemoryStorage(Storage): ...

Asynchronous API

Utilities and Parsing

Helper functions for parsing rate limit strings, managing window statistics, and handling dependencies for different storage backends.

def parse(limit_string: str) -> RateLimitItem: ...
def parse_many(limit_string: str) -> list[RateLimitItem]: ...

class WindowStats:
    reset_time: float
    remaining: int

Utilities

Types

Core type definitions used throughout the library:

from typing import NamedTuple

class WindowStats(NamedTuple):
    """Statistics for a rate limit window"""
    reset_time: float  # Time when window resets (seconds since epoch)
    remaining: int     # Remaining quota in current window

class Granularity(NamedTuple):
    """Time granularity definition"""
    seconds: int  # Duration in seconds
    name: str     # Granularity name (second, minute, hour, etc.)

Error Handling

The library defines several exception types for different error scenarios. These must be imported from limits.errors:

from limits.errors import ConfigurationError, ConcurrentUpdateError, StorageError
class ConfigurationError(Exception):
    """Raised when configuration problem is encountered"""

class ConcurrentUpdateError(Exception):
    """Raised when update fails due to concurrent updates"""
    def __init__(self, key: str, attempts: int):
        """
        Args:
            key: The key that failed to update
            attempts: Number of attempts made before giving up
        """

class StorageError(Exception):
    """Raised when error is encountered in storage backend"""
    def __init__(self, storage_error: Exception):
        """
        Args:
            storage_error: The underlying storage exception that occurred
        """
    storage_error: Exception  # The original exception that was wrapped