<img align="left" width="95" height="120" src="https://raw.githubusercontent.com/vutran1710/PyrateLimiter/master/docs/_static/logo.png">

PyrateLimiter

The request rate limiter using Leaky-bucket Algorithm.

Full project documentation can be found at pyratelimiter.readthedocs.io.

Upgrading from v3.x? See the Migration Guide for breaking changes.

<br>

Contents

• Features
• Installation
• Web Requests
  • AIOHTTP
  • HTTPX
  • Requests
• Quickstart
  • limiter_factory
  • Examples
• Basic usage
  • Key concepts
  • Defining rate limits & buckets
  • Defining clock & routing logic
  • Wrapping all up with Limiter
  • asyncio and event loops
  • as_decorator(): use limiter as decorator
  • Limiter API
    • Context Manager
  • Weight
  • Handling exceeded limits
    • Bucket analogy
    • Blocking vs Non-blocking
  • Backends
    • InMemoryBucket
    • MultiprocessBucket
    • SQLiteBucket
    • RedisBucket
    • PostgresBucket
    • BucketAsyncWrapper
  • Async or Sync or Multiprocessing
• Advanced Usage
  • Component-level Diagram
  • Time sources
  • Leaking
  • Concurrency
  • Custom backend

Features

• Supports unlimited rate limits and custom intervals.
• Separately tracks limits for different services or resources.
• Manages limit breaches with configurable blocking or non-blocking behavior.
• Offers multiple usage modes: direct calls or decorators.
• Fully compatible with both synchronous and asynchronous workflows.
• Provides SQLite and Redis backends for persistent limit tracking across threads or restarts.
• Includes MultiprocessBucket and SQLite File Lock backends for multiprocessing environments.

Installation

PyrateLimiter supports python ^3.8

Install using pip:

pip install pyrate-limiter

Or using conda:

conda install --channel conda-forge pyrate-limiter

Quickstart

To limit 5 requests within 2 seconds:

from pyrate_limiter import Duration, Rate, Limiter

limiter = Limiter(Rate(5, Duration.SECOND * 2))

# Blocking mode (default) - waits until permit available
for i in range(6):
    limiter.try_acquire(str(i))
    print(f"Acquired permit {i}")

# Non-blocking mode - returns False if bucket full
for i in range(6):
    success = limiter.try_acquire(str(i), blocking=False)
    if not success:
        print(f"Rate limited at {i}")

limiter_factory

limiter_factory.py provides several functions to simplify common cases:

• create_sqlite_limiter(rate_per_duration: int, duration: Duration, ...)
• create_inmemory_limiter(rate_per_duration: int, duration: Duration, ...)
• • more to be added...

Examples

• Rate limiting asyncio tasks: asyncio_ratelimit.py
• Rate limiting asyncio tasks w/ a decorator: asyncio_decorator.py
• HTTPX rate limiting - asyncio, single process and multiprocess examples httpx_ratelimiter.py
• Multiprocessing using an in-memory rate limiter - in_memory_multiprocess.py
• Multiprocessing using SQLite and a file lock - this can be used for distributed processes not created within a multiprocessing sql_filelock_multiprocess.py

Web Request Rate Limiting

pyrate_limiter provides three extras for popular web request libraries:

• AIOHTTP
• HTTPX
• Requests

AIOHTTP

from pyrate_limiter import 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)

Example: aiohttp_ratelimiter.py

HTTPX

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

limiter = limiter_factory.create_inmemory_limiter(rate_per_duration=1, duration=Duration.SECOND, max_delay=Duration.HOUR)
url = "https://example.com"

with httpx.Client(transport=RateLimiterTransport(limiter=limiter)) as client:
    client.get(url)

# or async
async with httpx.AsyncClient(transport=AsyncRateLimiterTransport(limiter=limiter)) as client:
    client.get(url)

...

Example: httpx_ratelimiter.py

Requests

from pyrate_limiter import 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)
....

Example: requests_ratelimiter.py

Basic Usage

Key concepts

Clock

• Timestamps incoming items

Bucket

• Stores items with timestamps.
• Functions as a FIFO queue.
• Can leak to remove outdated items.

BucketFactory

• Manages buckets and clocks, routing items to their appropriate buckets.
• Schedules periodic leak operations to prevent overflow.
• Allows custom logic for routing, conditions, and timing.

Limiter

• Provides a simple, intuitive API by abstracting underlying logic.
• Seamlessly supports both sync and async contexts.
• Offers multiple interaction modes: direct calls, decorators, and (future) context managers.
• Ensures thread-safety via RLock, and if needed, asyncio concurrency via asyncio.Lock

Defining rate limits and buckets

For example, an API (like LinkedIn or GitHub) might have these rate limits:

- 500 requests per hour
- 1000 requests per day
- 10000 requests per month

You can define these rates using the Rate class. Rate class has 2 properties only: limit and interval

from pyrate_limiter import Duration, Rate

hourly_rate = Rate(500, Duration.HOUR) # 500 requests per hour
daily_rate = Rate(1000, Duration.DAY) # 1000 requests per day
monthly_rate = Rate(10000, Duration.WEEK * 4) # 10000 requests per month

rates = [hourly_rate, daily_rate, monthly_rate]

Rates must be properly ordered:

• Rates' intervals & limits must be ordered from least to greatest
• Rates' ratio of limit/interval must be ordered from greatest to least

Buckets validate rates during initialization. If using a custom implementation, use the built-in validator:

from pyrate_limiter import validate_rate_list

assert validate_rate_list(my_rates)

Then, add the rates to the bucket of your choices

from pyrate_limiter import InMemoryBucket, RedisBucket

basic_bucket = InMemoryBucket(rates)

# Or, using redis
from redis import Redis

redis_connection = Redis(host='localhost')
redis_bucket = RedisBucket.init(rates, redis_connection, "my-bucket-name")

# Async Redis would work too!
from redis.asyncio import Redis

redis_connection = Redis(host='localhost')
redis_bucket = await RedisBucket.init(rates, redis_connection, "my-bucket-name")

If you only need a single Bucket for everything, and python's built-in time() is enough for you, then pass the bucket to Limiter then ready to roll!

from pyrate_limiter import Limiter

# Limiter constructor accepts single bucket as the only parameter,
# the rest are 3 optional parameters with default values as following
# Limiter(bucket, clock=MonotonicClock(), raise_when_fail=True, max_delay=None)
limiter = Limiter(bucket)

# Limiter is now ready to work!
limiter.try_acquire("hello world")

If you want to have finer grain control with routing & clocks etc, then you should use BucketFactory.

Defining Clock & routing logic with BucketFactory

When multiple bucket types are needed and items must be routed based on certain conditions, use BucketFactory.

First, define your