Blockbuster is a Python package designed to detect and prevent blocking calls within an asynchronous event loop. It's particularly useful during testing to ensure asynchronous code doesn't inadvertently perform blocking operations, which can cause performance bottlenecks. It works by monkey-patching common blocking functions and raising a `BlockingError` if called within an `asyncio` event loop. It currently only detects `asyncio` event loops and is tested with CPython.
pip install blockbusterVerified import paths — ran on the pinned version, not inferred.
This quickstart demonstrates how to use `blockbuster_ctx` to detect blocking calls within an `asyncio` event loop. It runs an `async` function that intentionally calls `time.sleep()`, which is a blocking operation. When `blockbuster_ctx` is active, this call will raise a `BlockingError`. The second part shows the same blocking call without `blockbuster` activated, which completes without error, highlighting the utility of the library.
Always pin to a specific minor version or range for stability, e.g., `pip install 'blockbuster<X.Y'` where X.Y is your current major.minor version.
Ensure your project exclusively uses `asyncio` or understand that other async contexts will not be monitored. For other frameworks, consider alternative detection mechanisms or custom rules.
Use Blockbuster predominantly with CPython. If using other Python implementations (e.g., PyPy, Jython), thoroughly test its behavior and be prepared for potential inconsistencies.
For such cases, you may need to add custom rules to Blockbuster's configuration during your test setup to specifically target the blocking functions used by those C-wrapped libraries. Contributions to the core project for common libraries are also welcome.
Replace the blocking call with its asynchronous equivalent (e.g., `await asyncio.sleep(1)`) or run the blocking operation in a separate thread using `await asyncio.to_thread(your_blocking_function, *args)`.
Ensure the correct casing is used for the class name: `from blockbuster import BlockBuster`.
Refactor the code to use asynchronous file I/O if available, wrap the blocking call in `await asyncio.to_thread(os.listdir, path)` to run it in a separate thread, or, if the blocking call is expected and safe, add a custom rule to `BlockBuster` to ignore it.
Use `asyncio`'s built-in asynchronous networking capabilities or wrap the blocking `socket.accept()` call with `await asyncio.to_thread(socket.accept)`.