| Name | blockbuster JSON |
| Version |
1.5.8
JSON |
| download |
| home_page | None |
| Summary | Utility to detect blocking calls in the async event loop |
| upload_time | 2024-12-24 08:18:35 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.8 |
| license | None |
| keywords |
async
asyncio
block
detect
event loop
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Blockbuster
Blockbuster is a Python package designed to detect and prevent blocking calls within an asynchronous event loop.
It is particularly useful when executing tests to ensure that your asynchronous code does not inadvertently call blocking operations,
which can lead to performance bottlenecks and unpredictable behavior.
In Python, the asynchronous event loop allows for concurrent execution of tasks without the need for multiple threads or processes.
This is achieved by running tasks cooperatively, where tasks yield control back to the event loop when they are waiting for I/O operations or other long-running tasks to complete.
However, blocking calls, such as file I/O operations or certain networking operations, can halt the entire event loop, preventing other tasks from running.
This can lead to increased latency and reduced performance, defeating the purpose of using asynchronous programming.
The difficulty with blocking calls is that they are not always obvious, especially when working with third-party libraries or legacy code.
This is where Blockbuster comes in: it helps you identify and eliminate blocking calls in your codebase during testing, ensuring that your asynchronous code runs smoothly and efficiently.
It does this by wrapping common blocking functions and raising an exception when they are called within an asynchronous context.
Notes:
- Blockbuster currently only detects `asyncio` event loops.
- Blockbuster is tested only with CPython. It may work with other Python implementations if it's possible to monkey-patch the functions with `setattr`.
## Installation
The package is named `blockbuster`.
For instance with `pip`:
```bash
pip install blockbuster
```
It is recommended to constrain the version of Blockbuster.
Blockbuster doesn't strictly follow semver. Breaking changes such as new rules added may be introduced between minor versions, but not between patch versions.
So it is recommended to constrain the Blockbuster version on the minor version.
For instance, with `uv`:
```bash
uv add "blockbuster>=1.5.5,<1.6"
```
## Using BlockBuster
### Manually
To activate BlockBuster manually, create an instance of the `BlockBuster` class and call the `activate()` method:
```python
from blockbuster import BlockBuster
blockbuster = BlockBuster()
blockbuster.activate()
```
Once activated, BlockBuster will raise a `BlockingError` exception whenever a blocking call is detected within an `asyncio` event loop.
To deactivate BlockBuster, call the `deactivate()` method:
```python
blockbuster.deactivate()
```
### Using the context manager
BlockBuster can also be activated using a context manager, which automatically activates and deactivates the checks within the `with` block:
```python
from blockbuster import blockbuster_ctx
with blockbuster_ctx():
# Your test code here
```
### Usage with Pytest
Blockbuster is intended to be used with testing frameworks like `pytest` to catch blocking calls.
Here's how you can integrate Blockbuster into your `pytest` test suite:
```python
import pytest
import time
from blockbuster import BlockBuster, BlockingError, blockbuster_ctx
from typing import Iterator
@pytest.fixture(autouse=True)
def blockbuster() -> Iterator[BlockBuster]:
with blockbuster_ctx() as bb:
yield bb
async def test_time_sleep() -> None:
with pytest.raises(BlockingError, match="sleep"):
time.sleep(1) # This should raise a BlockingError
```
By using the `blockbuster_ctx` context manager, Blockbuster is automatically activated for every test, and blocking calls will raise a `BlockingError`.
## How it works
Blockbuster works by wrapping common blocking functions from various modules (e.g., `os`, `socket`, `time`) and replacing them with versions that check if they are being called from within an `asyncio` event loop.
If such a call is detected, Blockbuster raises a `BlockingError` to indicate that a blocking operation is being performed inappropriately.
Blockbuster supports by default the following functions and modules:
- **Time Functions**:
- `time.sleep`
- **OS Functions**:
- `os.getcwd`
- `os.statvfs`
- `os.sendfile`
- `os.rename`
- `os.remove`
- `os.unlink`
- `os.mkdir`
- `os.rmdir`
- `os.link`
- `os.symlink`
- `os.readlink`
- `os.listdir`
- `os.scandir`
- `os.access`
- `os.stat`
- `os.replace`
- `os.read`
- `os.write`
- **OS path Functions**:
- `os.path.ismount`
- `os.path.samestat`
- `os.path.sameopenfile`
- `os.path.islink`
- `os.path.abspath`
- **IO Functions**:
- `io.BufferedReader.read`
- `io.BufferedWriter.write`
- `io.BufferedRandom.read`
- `io.BufferedRandom.write`
- `io.TextIOWrapper.read`
- `io.TextIOWrapper.write`
- **Socket Functions**:
- `socket.socket.connect`
- `socket.socket.accept`
- `socket.socket.send`
- `socket.socket.sendall`
- `socket.socket.sendto`
- `socket.socket.recv`
- `socket.socket.recv_into`
- `socket.socket.recvfrom`
- `socket.socket.recvfrom_into`
- `socket.socket.recvmsg`
- `ssl.SSLSocket.write`
- `ssl.SSLSocket.send`
- `ssl.SSLSocket.read`
- `ssl.SSLSocket.recv`
- **SQLite Functions**:
- `sqlite3.Cursor.execute`
- `sqlite3.Cursor.executemany`
- `sqlite3.Cursor.executescript`
- `sqlite3.Cursor.fetchone`
- `sqlite3.Cursor.fetchmany`
- `sqlite3.Cursor.fetchall`
- `sqlite3.Connection.execute`
- `sqlite3.Connection.executemany`
- `sqlite3.Connection.executescript`
- `sqlite3.Connection.commit`
- `sqlite3.Connection.rollback`
- **Thread lock Functions**:
- `threading.Lock.acquire`
- `threading.Lock.acquire_lock`
- **Built-in Functions**:
- `input`
Some exceptions to the rules are already in place:
- Importing modules does blocking calls as it interacts with the file system. Since this operation is cached and very hard to avoid, it is excluded from the detection.
- Blocking calls done by the `pydevd` debugger.
- Blocking calls done by the `pytest` framework.
## Customizing Blockbuster
### Adding custom rules
Blockbuster is not a silver bullet and may not catch all blocking calls.
In particular, it will not catch blocking calls that are done by third-party libraries that do blocking calls in C extensions.
For these third-party libraries, you can declare your own custom rules to Blockbuster to catch these blocking calls.
Eg.:
```python
from blockbuster import BlockBuster, BlockBusterFunction
import mymodule
blockbuster = BlockBuster()
blockbuster.functions["my_module.my_function"] = BlockBusterFunction(my_module, "my_function")
blockbuster.activate()
```
Note: if blockbuster has already been activated, you will need to activate the custom rule yourself.
```python
from blockbuster import blockbuster_ctx, BlockBusterFunction
import mymodule
with blockbuster_ctx() as blockbuster:
blockbuster.functions["my_module.my_function"] = BlockBusterFunction(my_module, "my_function")
blockbuster.functions["my_module.my_function"].activate()
```
### Allowing blocking calls in specific contexts
You can customize Blockbuster to allow blocking calls in specific functions by using the `can_block_in` method of the `BlockBusterFunction` class.
This method allows you to specify exceptions for particular files and functions where blocking calls are allowed.
```python
from blockbuster import BlockBuster
blockbuster = BlockBuster()
blockbuster.activate()
blockbuster.functions["os.stat"].can_block_in("specific_file.py", {"allowed_function"})
```
### Deactivating specific checks
If you need to deactivate specific checks, you can directly call the `deactivate` method on the corresponding `BlockBusterFunction` instance:
```python
from blockbuster import BlockBuster
blockbuster = BlockBuster()
blockbuster.activate()
blockbuster.functions["socket.socket.connect"].deactivate()
```
## Contributing
Contributions are welcome! If you encounter any issues or have suggestions for improvements, please open an issue on the GitHub repository.
### Development Setup
Blockbuster uses `uv` to manage its development environment.
See the [uv documentation](https://docs.astral.sh/uv/getting-started/installation/) for more informationand how to install it.
To install the required dependencies, run the following command:
```bash
uv sync
```
### Running Tests
Tests are written using `pytest`.
To run the tests, use the following command:
```bash
uv run pytest
```
### Code Formatting
Code formatting is done using `ruff`. To format the code, run the following command:
```bash
uv run ruff format
```
### Code Linting
Code linting is done using `ruff`.
To lint the code, fixing any issues that can be automatically fixed, run the following command:
```bash
uv run ruff check --fix
```
## License
This project is licensed under the Apache License 2.0 - see the [LICENSE](LICENSE) file for details.
## Acknowledgments
Blockbuster uses the [forbiddenfruit](https://clarete.li/forbiddenfruit/) library to monkey-patch CPython immutable builtin functions and methods.
Blockbuster was greatly inspired by the [BlockHound](https://github.com/reactor/BlockHound) library for Java, which serves a similar purpose of detecting blocking calls in JVM reactive applications.
Raw data
{
"_id": null,
"home_page": null,
"name": "blockbuster",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "async, asyncio, block, detect, event loop",
"author": null,
"author_email": "Christophe Bornet <bornet.chris@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/ef/d9/c117da06f9dee8ca4179e6a2fd3d413f6e4f7fcffd09652d2f2cda9d22da/blockbuster-1.5.8.tar.gz",
"platform": null,
"description": "# Blockbuster\n\nBlockbuster is a Python package designed to detect and prevent blocking calls within an asynchronous event loop.\nIt is particularly useful when executing tests to ensure that your asynchronous code does not inadvertently call blocking operations, \nwhich can lead to performance bottlenecks and unpredictable behavior.\n\nIn Python, the asynchronous event loop allows for concurrent execution of tasks without the need for multiple threads or processes.\nThis is achieved by running tasks cooperatively, where tasks yield control back to the event loop when they are waiting for I/O operations or other long-running tasks to complete.\n\nHowever, blocking calls, such as file I/O operations or certain networking operations, can halt the entire event loop, preventing other tasks from running.\nThis can lead to increased latency and reduced performance, defeating the purpose of using asynchronous programming.\n\nThe difficulty with blocking calls is that they are not always obvious, especially when working with third-party libraries or legacy code.\nThis is where Blockbuster comes in: it helps you identify and eliminate blocking calls in your codebase during testing, ensuring that your asynchronous code runs smoothly and efficiently.\nIt does this by wrapping common blocking functions and raising an exception when they are called within an asynchronous context.\n\nNotes:\n- Blockbuster currently only detects `asyncio` event loops.\n- Blockbuster is tested only with CPython. It may work with other Python implementations if it's possible to monkey-patch the functions with `setattr`.\n\n## Installation\n\nThe package is named `blockbuster`.\nFor instance with `pip`:\n\n```bash\npip install blockbuster\n```\n\nIt is recommended to constrain the version of Blockbuster.\nBlockbuster doesn't strictly follow semver. Breaking changes such as new rules added may be introduced between minor versions, but not between patch versions.\nSo it is recommended to constrain the Blockbuster version on the minor version.\nFor instance, with `uv`:\n\n```bash\nuv add \"blockbuster>=1.5.5,<1.6\"\n```\n\n## Using BlockBuster\n\n### Manually\n\nTo activate BlockBuster manually, create an instance of the `BlockBuster` class and call the `activate()` method:\n\n```python\nfrom blockbuster import BlockBuster\n\nblockbuster = BlockBuster()\nblockbuster.activate()\n```\n\nOnce activated, BlockBuster will raise a `BlockingError` exception whenever a blocking call is detected within an `asyncio` event loop.\n\nTo deactivate BlockBuster, call the `deactivate()` method:\n\n```python\nblockbuster.deactivate()\n```\n\n### Using the context manager\n\nBlockBuster can also be activated using a context manager, which automatically activates and deactivates the checks within the `with` block:\n\n```python\nfrom blockbuster import blockbuster_ctx\n\nwith blockbuster_ctx():\n # Your test code here\n```\n\n### Usage with Pytest\n\nBlockbuster is intended to be used with testing frameworks like `pytest` to catch blocking calls. \nHere's how you can integrate Blockbuster into your `pytest` test suite:\n\n```python\nimport pytest\nimport time\nfrom blockbuster import BlockBuster, BlockingError, blockbuster_ctx\nfrom typing import Iterator\n\n@pytest.fixture(autouse=True)\ndef blockbuster() -> Iterator[BlockBuster]:\n with blockbuster_ctx() as bb:\n yield bb\n\nasync def test_time_sleep() -> None:\n with pytest.raises(BlockingError, match=\"sleep\"):\n time.sleep(1) # This should raise a BlockingError\n```\n\nBy using the `blockbuster_ctx` context manager, Blockbuster is automatically activated for every test, and blocking calls will raise a `BlockingError`.\n\n## How it works\n\nBlockbuster works by wrapping common blocking functions from various modules (e.g., `os`, `socket`, `time`) and replacing them with versions that check if they are being called from within an `asyncio` event loop.\nIf such a call is detected, Blockbuster raises a `BlockingError` to indicate that a blocking operation is being performed inappropriately.\n\nBlockbuster supports by default the following functions and modules:\n\n- **Time Functions**:\n - `time.sleep`\n- **OS Functions**:\n - `os.getcwd`\n - `os.statvfs`\n - `os.sendfile`\n - `os.rename`\n - `os.remove`\n - `os.unlink`\n - `os.mkdir`\n - `os.rmdir`\n - `os.link`\n - `os.symlink`\n - `os.readlink`\n - `os.listdir`\n - `os.scandir`\n - `os.access`\n - `os.stat`\n - `os.replace`\n - `os.read`\n - `os.write`\n- **OS path Functions**:\n - `os.path.ismount`\n - `os.path.samestat`\n - `os.path.sameopenfile`\n - `os.path.islink`\n - `os.path.abspath`\n- **IO Functions**:\n - `io.BufferedReader.read`\n - `io.BufferedWriter.write`\n - `io.BufferedRandom.read`\n - `io.BufferedRandom.write`\n - `io.TextIOWrapper.read`\n - `io.TextIOWrapper.write`\n- **Socket Functions**:\n - `socket.socket.connect`\n - `socket.socket.accept`\n - `socket.socket.send`\n - `socket.socket.sendall`\n - `socket.socket.sendto`\n - `socket.socket.recv`\n - `socket.socket.recv_into`\n - `socket.socket.recvfrom`\n - `socket.socket.recvfrom_into`\n - `socket.socket.recvmsg`\n - `ssl.SSLSocket.write`\n - `ssl.SSLSocket.send`\n - `ssl.SSLSocket.read`\n - `ssl.SSLSocket.recv`\n- **SQLite Functions**:\n - `sqlite3.Cursor.execute`\n - `sqlite3.Cursor.executemany`\n - `sqlite3.Cursor.executescript`\n - `sqlite3.Cursor.fetchone`\n - `sqlite3.Cursor.fetchmany`\n - `sqlite3.Cursor.fetchall`\n - `sqlite3.Connection.execute`\n - `sqlite3.Connection.executemany`\n - `sqlite3.Connection.executescript`\n - `sqlite3.Connection.commit`\n - `sqlite3.Connection.rollback`\n- **Thread lock Functions**:\n - `threading.Lock.acquire`\n - `threading.Lock.acquire_lock`\n- **Built-in Functions**:\n - `input`\n\nSome exceptions to the rules are already in place:\n- Importing modules does blocking calls as it interacts with the file system. Since this operation is cached and very hard to avoid, it is excluded from the detection.\n- Blocking calls done by the `pydevd` debugger.\n- Blocking calls done by the `pytest` framework.\n\n## Customizing Blockbuster\n\n### Adding custom rules\n\nBlockbuster is not a silver bullet and may not catch all blocking calls.\nIn particular, it will not catch blocking calls that are done by third-party libraries that do blocking calls in C extensions.\nFor these third-party libraries, you can declare your own custom rules to Blockbuster to catch these blocking calls.\n\nEg.:\n```python\nfrom blockbuster import BlockBuster, BlockBusterFunction\nimport mymodule\n\nblockbuster = BlockBuster()\nblockbuster.functions[\"my_module.my_function\"] = BlockBusterFunction(my_module, \"my_function\")\nblockbuster.activate()\n```\n\nNote: if blockbuster has already been activated, you will need to activate the custom rule yourself.\n\n```python\nfrom blockbuster import blockbuster_ctx, BlockBusterFunction\nimport mymodule\n\nwith blockbuster_ctx() as blockbuster:\n blockbuster.functions[\"my_module.my_function\"] = BlockBusterFunction(my_module, \"my_function\")\n blockbuster.functions[\"my_module.my_function\"].activate()\n```\n\n### Allowing blocking calls in specific contexts\n\nYou can customize Blockbuster to allow blocking calls in specific functions by using the `can_block_in` method of the `BlockBusterFunction` class.\nThis method allows you to specify exceptions for particular files and functions where blocking calls are allowed.\n\n```python\nfrom blockbuster import BlockBuster\n\nblockbuster = BlockBuster()\nblockbuster.activate()\nblockbuster.functions[\"os.stat\"].can_block_in(\"specific_file.py\", {\"allowed_function\"})\n```\n\n### Deactivating specific checks\n\nIf you need to deactivate specific checks, you can directly call the `deactivate` method on the corresponding `BlockBusterFunction` instance:\n\n```python\nfrom blockbuster import BlockBuster\n\nblockbuster = BlockBuster()\nblockbuster.activate()\nblockbuster.functions[\"socket.socket.connect\"].deactivate()\n```\n\n## Contributing\n\nContributions are welcome! If you encounter any issues or have suggestions for improvements, please open an issue on the GitHub repository.\n\n### Development Setup\n\nBlockbuster uses `uv` to manage its development environment.\nSee the [uv documentation](https://docs.astral.sh/uv/getting-started/installation/) for more informationand how to install it.\n\nTo install the required dependencies, run the following command:\n\n```bash\nuv sync\n```\n\n### Running Tests\n\nTests are written using `pytest`.\nTo run the tests, use the following command:\n\n```bash\nuv run pytest\n```\n\n### Code Formatting\n\nCode formatting is done using `ruff`. To format the code, run the following command:\n\n```bash\nuv run ruff format\n```\n\n### Code Linting\n\nCode linting is done using `ruff`.\nTo lint the code, fixing any issues that can be automatically fixed, run the following command:\n\n```bash\nuv run ruff check --fix\n```\n\n## License\n\nThis project is licensed under the Apache License 2.0 - see the [LICENSE](LICENSE) file for details.\n\n## Acknowledgments\n\nBlockbuster uses the [forbiddenfruit](https://clarete.li/forbiddenfruit/) library to monkey-patch CPython immutable builtin functions and methods.\n\nBlockbuster was greatly inspired by the [BlockHound](https://github.com/reactor/BlockHound) library for Java, which serves a similar purpose of detecting blocking calls in JVM reactive applications.\n",
"bugtrack_url": null,
"license": null,
"summary": "Utility to detect blocking calls in the async event loop",
"version": "1.5.8",
"project_urls": {
"Issues": "https://github.com/cbornet/blockbuster/issues",
"Repository": "https://github.com/cbornet/blockbuster.git"
},
"split_keywords": [
"async",
" asyncio",
" block",
" detect",
" event loop"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "1a611c154c6f4e2994d7b1a9fa9cc40d7614c9fc0fe07cc55c1e1ac956404033",
"md5": "cd57cefe7cb374c1ee0b42e993941e09",
"sha256": "ea0352823acbd837872785a96ec87701f1c7938410d66c6a8857ae11646d9744"
},
"downloads": -1,
"filename": "blockbuster-1.5.8-py3-none-any.whl",
"has_sig": false,
"md5_digest": "cd57cefe7cb374c1ee0b42e993941e09",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 11700,
"upload_time": "2024-12-24T08:18:31",
"upload_time_iso_8601": "2024-12-24T08:18:31.207877Z",
"url": "https://files.pythonhosted.org/packages/1a/61/1c154c6f4e2994d7b1a9fa9cc40d7614c9fc0fe07cc55c1e1ac956404033/blockbuster-1.5.8-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "efd9c117da06f9dee8ca4179e6a2fd3d413f6e4f7fcffd09652d2f2cda9d22da",
"md5": "8632abf8ca61aaec10894aa5b043ec63",
"sha256": "0018080fd735e84f0b138f15ff0a339f99444ecdc0045f16056c7b23db92706b"
},
"downloads": -1,
"filename": "blockbuster-1.5.8.tar.gz",
"has_sig": false,
"md5_digest": "8632abf8ca61aaec10894aa5b043ec63",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 33990,
"upload_time": "2024-12-24T08:18:35",
"upload_time_iso_8601": "2024-12-24T08:18:35.284031Z",
"url": "https://files.pythonhosted.org/packages/ef/d9/c117da06f9dee8ca4179e6a2fd3d413f6e4f7fcffd09652d2f2cda9d22da/blockbuster-1.5.8.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-12-24 08:18:35",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "cbornet",
"github_project": "blockbuster",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "blockbuster"
}
