aiolimiter

By default, AsyncLimiter allows for 'bursting'. If capacity has not been used for a while, multiple requests can enter the limited section in quick succession until the capacity is exhausted. The maximum burst size is equal to the max_rate value.

To disable bursting and enforce a strict interval between entries, set the max_rate to 1 and set the time_period to your desired minimum interval.

# Allow bursting (default behavior)
# If max_rate is 4, the first 4 tasks can run immediately
limiter = AsyncLimiter(4, 8)

# No bursts: allow entry exactly every 1.5 seconds
limiter = AsyncLimiter(1, 1.5)

Install the aiolimiter package using pip. This library requires Python 3.10 or newer.

$ pip install aiolimiter

The most common way to use aiolimiter is to wrap a code section with async with. This ensures the section is entered only within the specified rate limits.

Important: Create AsyncLimiter instances per asyncio loop. Re-using a limiter across different asyncio loops is not supported and can lead to undefined behavior.

from aiolimiter import AsyncLimiter

async def main():
    # Limits to 100 entries per minute
    limiter = AsyncLimiter(100)

    async with limiter:
        # This section is rate-limited
        pass

The AsyncLimiter class implements the Leaky Bucket algorithm to control the rate at which a code section is entered. You can use it as an asynchronous context manager to ensure that a specific number of entries occur within a defined time window.

from aiolimiter import AsyncLimiter

# allow for 100 concurrent entries within a 30 second window
rate_limit = AsyncLimiter(100, 30)

async def some_coroutine():
    async with rate_limit:
        # this section is *at most* going to entered 100 times
        # in a 30 second period.
        await do_something()

In addition to the context manager, you can interact with the limiter using these methods:

• await limiter.acquire(n=1): Blocks until n amount of capacity is available.
• limiter.has_capacity(n=1): Returns True if there is enough capacity for n, otherwise False. Use this to reject requests immediately instead of waiting.

# Block until capacity is available
await limiter.acquire()

# Check capacity without blocking
if limiter.has_capacity():
    # proceed
else:
    # reject request

You can treat some requests as 'heavier' or 'lighter' by passing a value to acquire() or has_capacity().

Warning: When mixing capacity amounts, small requests tend to be prioritized over large ones when the limiter is near its maximum rate, because small requests are more likely to find enough immediate free capacity.

# Acquire a large block of capacity
await limit.acquire(10)

# Check for a small amount of capacity
if limit.has_capacity(0.5):
    pass

The AsyncLimiter class is the primary interface for aiolimiter. It is used to implement rate limiting in asynchronous Python applications, typically following a leaky bucket algorithm to control the rate of concurrent or periodic operations.

from aiolimiter import AsyncLimiter

# Example usage (conceptual):
# limiter = AsyncLimiter(max_rate, period)
# async with limiter:
#     await do_something()