{"id":3889,"library":"asyncio-throttle","title":"asyncio-throttle","description":"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.","status":"active","version":"1.0.2","language":"en","source_language":"en","source_url":"https://github.com/hallazzang/asyncio-throttle","tags":["async","asyncio","throttle","rate-limit","concurrency"],"install":[{"cmd":"pip install asyncio-throttle","lang":"bash","label":"Install stable version"}],"dependencies":[],"imports":[{"symbol":"Throttler","correct":"from asyncio_throttle import Throttler"}],"quickstart":{"code":"import asyncio\nimport time\nimport random\nfrom asyncio_throttle import Throttler\n\nasync def worker(no, throttler, n):\n    for _ in range(n):\n        # Simulate some async work before acquiring the throttle\n        await asyncio.sleep(random.random() * 0.5)\n        async with throttler:\n            # This block will be rate-limited\n            print(f'{time.time():.4f} Worker #{no}: Bang!')\n\nasync def main():\n    # Limit to 5 'Bang!' messages per second across all workers\n    throttler = Throttler(rate_limit=5)\n    \n    # Create 5 workers, each trying to 'Bang!' 3 times\n    tasks = [\n        asyncio.create_task(worker(no, throttler, 3))\n        for no in range(5)\n    ]\n    await asyncio.gather(*tasks)\n\nif __name__ == '__main__':\n    # Ensure to run in an event loop\n    asyncio.run(main())","lang":"python","description":"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."},"warnings":[{"fix":"Ensure all coroutine calls are `await`ed or scheduled via `asyncio.create_task()`.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"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.","message":"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()`.","severity":"gotcha","affected_versions":"All versions"},{"fix":"Offload CPU-bound work to `asyncio.to_thread()` or `loop.run_in_executor(None, blocking_function, *args)`.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"fix":"Use explicit `await` for all async function calls in affected environments. Avoid `asyncio.run()` in already running event loops.","message":"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`.","severity":"breaking","affected_versions":"Python 3.14+"}],"env_vars":null,"last_verified":"2026-04-11T00:00:00.000Z","next_check":"2026-07-10T00:00:00.000Z"}