asyncio-throttle

1.0.2 · active · verified Sat Apr 11

asyncio-throttle is a simple, easy-to-use library providing a throttler for asynchronous operations in Python. It's designed to limit the rate at which code blocks can be executed, primarily through an `async with` statement. The current stable version is 1.0.2, released in April 2021. It requires Python 3.6 or later.

Warnings

gotcha Forgetting to `await` a coroutine function will create a coroutine object but not schedule or run it, leading to silent failures or unexpected behavior. Always use `await` or `asyncio.create_task()` for coroutines.
Fix: Ensure all coroutine calls are `await`ed or scheduled via `asyncio.create_task()`.
gotcha Performing blocking I/O operations (e.g., `requests.get()`, `time.sleep()`) directly within an `asyncio` event loop will block the entire loop, negating the benefits of asynchronous programming. Use async-native libraries (e.g., `aiohttp`, `httpx`, `asyncpg`) or offload blocking operations to a thread pool via `loop.run_in_executor()`.
Fix: Replace blocking I/O with async equivalents or use `asyncio.to_thread()` (Python 3.9+) / `loop.run_in_executor()` for CPU-bound or blocking tasks.
gotcha CPU-bound tasks executed directly in the event loop will block all other concurrent tasks. For computationally intensive work, use `loop.run_in_executor()` or `asyncio.to_thread()` (Python 3.9+) to run them in a separate thread or process pool, preventing the event loop from stalling.
Fix: Offload CPU-bound work to `asyncio.to_thread()` or `loop.run_in_executor(None, blocking_function, *args)`.
breaking Changes in `asyncio` implementation in Python 3.14+ may affect environments that relied on `nest_asyncio` for automatic async-to-sync conversion (e.g., Jupyter notebooks). Direct `await` calls for async methods are now often required, and implicit conversions may fail with `RuntimeError`.
Fix: Use explicit `await` for all async function calls in affected environments. Avoid `asyncio.run()` in already running event loops.

Install

pip install asyncio-throttle Install stable version

Imports

Throttler
```
from asyncio_throttle import Throttler
```

Quickstart

This quickstart demonstrates how to use `Throttler` as an asynchronous context manager to limit the rate of operations across multiple concurrent tasks. It creates five workers that each attempt to print a message, but the `Throttler` ensures a maximum of 5 messages are printed per second in total.

import asyncio
import time
import random
from asyncio_throttle import Throttler

async def worker(no, throttler, n):
    for _ in range(n):
        # Simulate some async work before acquiring the throttle
        await asyncio.sleep(random.random() * 0.5)
        async with throttler:
            # This block will be rate-limited
            print(f'{time.time():.4f} Worker #{no}: Bang!')

async def main():
    # Limit to 5 'Bang!' messages per second across all workers
    throttler = Throttler(rate_limit=5)
    
    # Create 5 workers, each trying to 'Bang!' 3 times
    tasks = [
        asyncio.create_task(worker(no, throttler, 3))
        for no in range(5)
    ]
    await asyncio.gather(*tasks)

if __name__ == '__main__':
    # Ensure to run in an event loop
    asyncio.run(main())

view raw JSON →