Utility to detect blocking calls in the async event loop

1.5.26 · active · verified Thu Apr 09

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.

Warnings

breaking Breaking changes, such as new rules, may be introduced between minor versions. It is recommended to constrain the Blockbuster version on the minor version (e.g., `blockbuster>=1.5.0,<1.6`).
Fix: 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.
gotcha Blockbuster currently only detects `asyncio` event loops. It does not provide detection for other asynchronous frameworks.
Fix: 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.
gotcha Blockbuster is primarily tested with CPython. While it might work with other Python implementations, its functionality relies on the ability to monkey-patch functions with `setattr`, which may not be consistently available or behave identically across all interpreters.
Fix: Use Blockbuster predominantly with CPython. If using other Python implementations (e.g., PyPy, Jython), thoroughly test its behavior and be prepared for potential inconsistencies.
gotcha Blockbuster may not detect blocking calls made by third-party libraries that wrap C libraries instead of using standard Python framework methods for I/O.
Fix: 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.

Install

pip install blockbuster Install latest version

Imports

blockbuster_ctx
```
from blockbuster import blockbuster_ctx
```
Context manager to activate blocking call detection.
BlockingError
```
from blockbuster import BlockingError
```
The exception raised when a blocking call is detected.

Quickstart

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.

import asyncio
import time
from blockbuster import blockbuster_ctx, BlockingError

async def main():
    print("Running async task with blockbuster (should fail on time.sleep)...")
    try:
        with blockbuster_ctx():
            # This is a blocking call and should raise BlockingError
            time.sleep(0.1)
        print("This line should not be reached if blocking call is detected.")
    except BlockingError as e:
        print(f"Caught expected BlockingError: {e}")
    except Exception as e:
        print(f"Caught unexpected error: {e}")

    print("\nRunning async task without blockbuster (should complete)...")
    try:
        # This will complete without error
        time.sleep(0.1)
        print("Blocking call completed without blockbuster activated.")
    except BlockingError as e:
        print(f"Caught BlockingError (unexpected): {e}")

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

view raw JSON →