DOCSEARCHbyAlgolia
Submit Docs

pyrate-limiter

repository·master·Indexed 19 days ago

https://github.com/vutran1710/pyratelimiter

A fast, async-friendly Python rate limiter implementing the Leaky-Bucket algorithm. It supports multiple simultaneous rates, per-key limits, and variable request costs via weights. The library provides various pluggable backends including InMemoryBucket, RedisBucket, SQLiteBucket, PostgresBucket, and MultiprocessBucket. It includes built-in support for asyncio and provides integration helpers for HTTP clients such as aiohttp, httpx, and requests.

Tokens
10.3K
Snippets
39
Records
50
Agent score
67%

What's inside pyrate-limiter

  1. How to use custom or distributed clocks

    master

    In v4, each bucket owns its time source via bucket.now(). The Limiter no longer takes a clock parameter. To ensure distributed workers agree on time, you have two options:

    1. Override now() on a bucket subclass: This is the recommended approach as it keeps leak consistent.
    2. Inject a clock into a bucket: For buckets that support self._clock (like InMemoryBucket or PostgresBucket), you can assign a custom clock instance to that attribute.

    Built-in clocks: MonotonicClock (default), MonotonicAsyncClock, PostgresClock, SQLiteClock.

    Example: Overriding now() for Redis-based time

    class RedisTimeBucket(RedisBucket):
    def now(self) -> int:
        seconds, microseconds = self.redis.time()
        return seconds * 1000 + microseconds // 1000
    from pyrate_limiter import AbstractClock, InMemoryBucket, RedisBucket, Rate, Duration

class RedisClock(AbstractClock):
    def __init__(self, redis):
        self.redis = redis

    def now(self) -> int:
        seconds, microseconds = self.redis.time()
        return seconds * 1000 + microseconds // 1000

# Option A — override now() (recommended)
class RedisTimeBucket(RedisBucket):
    def now(self) -> int:
        seconds, microseconds = self.redis.time()
        return seconds * 1000 + microseconds // 1000

# Option B — inject a clock into a bucket that uses self._clock
bucket = InMemoryBucket([Rate(5, Duration.SECOND)])
bucket._clock = RedisClock(redis_client)

  2. Choose a Bucket Backend

    master

    Pyrate-limiter supports several backends depending on your concurrency and persistence requirements. Every bucket accepts a List[Rate].

    BackendSyncAsyncPersistentMulti-processBest for
    InMemoryBucket✅(wrap)❌❌single process, fastest
    SQLiteBucket✅❌✅✅ (file lock)persistence / one host, many processes
    RedisBucket✅✅✅✅distributed across hosts
    PostgresBucket✅❌✅✅distributed, already on Postgres
    MultiprocessBucket✅(wrap)❌✅a single multiprocessing pool
    BucketAsyncWrapper—✅——make any sync bucket async-safe

  3. How PyrateLimiter works: Core Concepts

    master

    PyrateLimiter implements the Leaky-Bucket algorithm. A bucket represents a fixed capacity; it fills as requests arrive and 'leaks' (removes expired items) at a constant rate. When the bucket is full, new requests are either delayed or rejected.

    Key abstractions:

    • Clock: Provides timestamps (now() -> int).
    • Bucket: Stores timestamped items and enforces rates by leaking expired items.
    • BucketFactory: Routes items to the correct bucket based on a key and manages background leaking.
    • Limiter: The primary public API (façade) that provides sync/async, blocking/non-blocking, and decorator interfaces.

  4. Quickstart: Basic Rate Limiting

    master

    To limit requests to a specific rate (e.g., 5 requests per 2 seconds) using an in-memory bucket, use the Limiter class.

    By default, try_acquire is blocking, meaning it will wait until a permit is available. To fail fast without waiting, set blocking=False.

    from pyrate_limiter import Duration, Rate, Limiter

# A Limiter with a single rate, backed by an in-memory bucket
limiter = Limiter(Rate(5, Duration.SECOND * 2))

# Blocking (default): waits until a permit is available
for i in range(6):
    limiter.try_acquire("my-resource")
    print(f"acquired {i}")

# Non-blocking: returns False immediately when the bucket is full
if not limiter.try_acquire("my-resource", blocking=False):
    print("rate limited!")

  5. Quickstart: Using limiter_factory

    master

    For common use cases, limiter_factory provides a convenient one-liner to create an in-memory limiter.

    from pyrate_limiter import Duration, limiter_factory

limiter = limiter_factory.create_inmemory_limiter(rate_per_duration=5, duration=Duration.SECOND)
limiter.try_acquire("my-resource")

  6. Migrate from PyrateLimiter 3.x to 4.0

    master

    When upgrading to version 4.0, several breaking changes affect how rate limiting is handled. The most significant shift is moving from exception-based flow control to a blocking/non-blocking model.

    Key changes include:

    • try_acquire now blocks by default instead of raising BucketFullException.
    • Proper async support is provided via try_acquire_async.
    • The decorator API is simplified and no longer requires a mapping function.
    • BucketFullException and LimiterDelayException have been removed.
    • The Limiter constructor is simplified as clock responsibility has moved to the Bucket.

  7. Rate limit HTTP requests with extras

    master

    The pyrate_limiter.extras module provides drop-in helpers for popular HTTP clients.

    AIOHTTP

    from pyrate_limiter import Duration, limiter_factory
from pyrate_limiter.extras.aiohttp_limiter import RateLimitedSession

limiter = limiter_factory.create_inmemory_limiter(rate_per_duration=2, duration=Duration.SECOND)
session = RateLimitedSession(limiter)

    HTTPX

    import httpx
from pyrate_limiter import Duration, limiter_factory
from pyrate_limiter.extras.httpx_limiter import AsyncRateLimiterTransport, RateLimiterTransport

limiter = limiter_factory.create_inmemory_limiter(rate_per_duration=1, duration=Duration.SECOND)

# Sync
with httpx.Client(transport=RateLimiterTransport(limiter=limiter)) as client:
    client.get("https://example.com")

# Async
async with httpx.AsyncClient(transport=AsyncRateLimiterTransport(limiter=limiter)) as client:
    await client.get("https://example.com")

    Requests

    from pyrate_limiter import Duration, limiter_factory
from pyrate_limiter.extras.requests_limiter import RateLimitedRequestsSession

limiter = limiter_factory.create_inmemory_limiter(rate_per_duration=2, duration=Duration.SECOND)
session = RateLimitedRequestsSession(limiter)

  8. Install PyrateLimiter

    master

    PyrateLimiter requires Python 3.10+. You can install it via pip or conda.

    To install the core package:

    pip install pyrate-limiter
# or
conda install --channel conda-forge pyrate-limiter

    To install with all optional backend drivers (Redis, Postgres, and filelock):

    pip install "pyrate-limiter[all]"
    pip install pyrate-limiter

  9. Implement custom routing with BucketFactory

    master

    To route items to different buckets (e.g., per user or per endpoint), implement a BucketFactory. You must define wrap_item and get.

    To create buckets on demand, use self.create(bucket_class, *args, **kwargs), which builds the bucket and automatically schedules its background leaking.

    Example: Per-name bucket routing

    from pyrate_limiter import ()
    AbstractBucket, BucketFactory, RateItem, MonotonicClock,
    InMemoryBucket, Rate, Duration, Limiter,
)

class PerNameFactory(BucketFactory):
    def __init__(self, clock):
        self.clock = clock
        self.buckets = {}

    def wrap_item(self, name: str, weight: int = 1) -> RateItem:
        return RateItem(name, self.clock.now(), weight=weight)

    def get(self, item: RateItem) -> AbstractBucket:
        if item.name not in self.buckets:
            # create() builds the bucket AND schedules its leak
            self.buckets[item.name] = self.create(InMemoryBucket, [Rate(5, Duration.SECOND)])
        return self.buckets[item.name]

# Usage
limiter = Limiter(PerNameFactory(MonotonicClock()))
limiter.try_acquire("user-123")
    from pyrate_limiter import (
    AbstractBucket, BucketFactory, RateItem, MonotonicClock,
    InMemoryBucket, Rate, Duration, Limiter,
)

class PerNameFactory(BucketFactory):
    def __init__(self, clock):
        self.clock = clock
        self.buckets = {}

    def wrap_item(self, name: str, weight: int = 1) -> RateItem:
        return RateItem(name, self.clock.now(), weight=weight)

    def get(self, item: RateItem) -> AbstractBucket:
        if item.name not in self.buckets:
            self.buckets[item.name] = self.create(InMemoryBucket, [Rate(5, Duration.SECOND)])
        return self.buckets[item.name]

limiter = Limiter(PerNameFactory(MonotonicClock()))
limiter.try_acquire("user-123")

  10. Implement time tracking with AbstractClock

    master

    The AbstractClock is the base interface for all clock implementations in pyrate-limiter. It defines a standard now() method that returns the current time in milliseconds. Depending on the implementation, this may return a synchronous int or an Awaitable[int].

    from pyrate_limiter.clocks import AbstractClock

# Implement your own clock if needed
class MyCustomClock(AbstractClock):
    def now(self) -> int:
        return 123456789  # Example timestamp in ms

  11. Configure the Limiter constructor

    master

    The Limiter constructor in 4.0 is more streamlined. Many parameters related to clock management and exception handling have been removed because they are now handled by the Bucket or controlled via the try_acquire call parameters.

    Removed parameters:

    • clock: Buckets now manage their own clock via bucket.now().
    • raise_when_fail: Use blocking=False instead.
    • max_delay: Controlled per-call via the blocking parameter.
    • retry_until_max_delay: Blocking mode retries automatically.
    limiter = Limiter(
    bucket,
    buffer_ms=50,  # optional, default 50ms
)

  12. Initialize the Limiter

    master

    The Limiter class is the primary entry point for managing rate limits. You can initialize it using a single Rate, a list of Rate objects, an AbstractBucket, or a BucketFactory.

    If you provide a list of Rate objects, the limiter automatically initializes an InMemoryBucket with those rates.

    If you provide an AbstractBucket, it is wrapped in a SingleBucketFactory.

    from pyrate_limiter import Limiter, Rate

# Using a single rate
limiter = Limiter(Rate(10, 60))

# Using multiple rates
limiter = Limiter([Rate(10, 60), Rate(100, 3600)])

# Using an existing bucket
# limiter = Limiter(my_bucket)