asynciolimiter

1.2.0 · active · verified Thu Apr 16

asynciolimiter is a simple yet efficient Python library providing various rate limiting algorithms for AsyncIO applications. It offers different limiter flavors like `Limiter`, `LeakyBucketLimiter`, and `StrictLimiter` to control the rate of asynchronous operations. The current version is 1.2.0, actively maintained with regular releases addressing features and bug fixes.

Common errors

```
ModuleNotFoundError: No module named 'asynciolimiter.AsyncLimiter'
```
cause Attempting to import `AsyncLimiter`, which belongs to a different library (`aiolimiter`), instead of `asynciolimiter`'s own limiter classes.

fix If you intend to use `asynciolimiter`, import its specific limiter classes, such as `Limiter`, `LeakyBucketLimiter`, or `StrictLimiter`. For example: `from asynciolimiter import Limiter`.
```
LeakyBucketLimiter does not release tasks after periods of inactivity, or tasks are stuck waiting indefinitely.
```
cause This behavior was caused by a bug in older versions of `LeakyBucketLimiter` where it failed to schedule tasks correctly after its bucket had fully emptied.

fix Ensure you are using `asynciolimiter` version `1.1.2` or higher. Upgrade using `pip install --upgrade asynciolimiter`.
```
My asynchronous tasks are being executed too slowly (or too quickly) despite setting a rate limit.
```
cause The `rate` parameter was likely set incorrectly. It expects the number of calls per second, not total calls over a period.

fix Recalculate your desired rate as `total_calls / total_seconds`. For example, if you want 30 calls in 60 seconds, your rate should be `30/60 = 0.5`. Initialize the limiter as `Limiter(0.5)`.

Warnings

breaking `LeakyBucketLimiter` in versions prior to 1.1.2 had a bug where it would not schedule tasks correctly after the bucket fully emptied.
Fix: Upgrade to `asynciolimiter>=1.1.2` to ensure correct behavior of `LeakyBucketLimiter`.
gotcha In versions prior to 1.1.1, the library might raise an assertion error if the event loop was too fast, which has since been downgraded to a warning.
Fix: Upgrade to `asynciolimiter>=1.1.1` to avoid assertions and instead receive a warning for fast event loop scenarios.
gotcha The `rate` parameter for all limiters (e.g., `Limiter`, `LeakyBucketLimiter`, `StrictLimiter`) expects calls *per second*. Misinterpreting this can lead to unexpected rate limiting behavior (e.g., too slow or too fast).
Fix: Always pass the desired rate as a floating-point number representing calls per second. For example, to allow 10 requests in 5 seconds, use `Limiter(10/5)`, which evaluates to `Limiter(2.0)`.

Install

pip install asynciolimiter Install stable version

Imports

Limiter
```
from asynciolimiter import Limiter
```

LeakyBucketLimiter

from asynciolimiter import LeakyBucketLimiter

StrictLimiter

from asynciolimiter import StrictLimiter

Quickstart

This quickstart demonstrates how to use the basic `Limiter` to constrain the execution rate of asynchronous tasks. It creates a limiter allowing 2 requests per second (10 requests over 5 seconds) and then schedules 10 tasks, each waiting for the limiter before proceeding.

import asyncio
from asynciolimiter import Limiter

# Limit to 2 requests per second (10 requests per 5 seconds)
rate_limiter = Limiter(10/5)

async def request_task(task_id):
    await rate_limiter.wait() # Wait for a slot to be available
    print(f"Task {task_id}: Request sent at {asyncio.get_event_loop().time():.2f}")

async def main():
    print("Starting rate-limited tasks...")
    tasks = [request_task(i) for i in range(10)]
    await asyncio.gather(*tasks)
    print("All tasks completed.")

if __name__ == "__main__":
    asyncio.run(main())

view raw JSON →