
Security News
npm ‘is’ Package Hijacked in Expanding Supply Chain Attack
The ongoing npm phishing campaign escalates as attackers hijack the popular 'is' package, embedding malware in multiple versions.
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:
asyncio
event loops.setattr
.The package is named blockbuster
.
For instance with pip
:
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
:
uv add "blockbuster>=1.5.5,<1.6"
To activate BlockBuster manually, create an instance of the BlockBuster
class and call the activate()
method:
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:
blockbuster.deactivate()
BlockBuster can also be activated using a context manager, which automatically activates and deactivates the checks within the with
block:
from blockbuster import blockbuster_ctx
with blockbuster_ctx():
# Your test code here
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:
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
.
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.sleep
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.ismount
os.path.samestat
os.path.sameopenfile
os.path.islink
os.path.abspath
io.BufferedReader.read
io.BufferedWriter.write
io.BufferedRandom.read
io.BufferedRandom.write
io.TextIOWrapper.read
io.TextIOWrapper.write
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
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
threading.Lock.acquire
threading.Lock.acquire_lock
input
Some exceptions to the rules are already in place:
pydevd
debugger.pytest
framework.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.:
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.
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()
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.
from blockbuster import BlockBuster
blockbuster = BlockBuster()
blockbuster.activate()
blockbuster.functions["os.stat"].can_block_in("specific_file.py", {"allowed_function"})
If you need to deactivate specific checks, you can directly call the deactivate
method on the corresponding BlockBusterFunction
instance:
from blockbuster import BlockBuster
blockbuster = BlockBuster()
blockbuster.activate()
blockbuster.functions["socket.socket.connect"].deactivate()
Contributions are welcome! If you encounter any issues or have suggestions for improvements, please open an issue on the GitHub repository.
Blockbuster uses uv
to manage its development environment.
See the uv documentation for more informationand how to install it.
To install the required dependencies, run the following command:
uv sync
Tests are written using pytest
.
To run the tests, use the following command:
uv run pytest
Code formatting is done using ruff
. To format the code, run the following command:
uv run ruff format
Code linting is done using ruff
.
To lint the code, fixing any issues that can be automatically fixed, run the following command:
uv run ruff check --fix
This project is licensed under the Apache License 2.0 - see the LICENSE file for details.
Blockbuster uses the forbiddenfruit library to monkey-patch CPython immutable builtin functions and methods.
Blockbuster was greatly inspired by the BlockHound library for Java, which serves a similar purpose of detecting blocking calls in JVM reactive applications.
FAQs
Utility to detect blocking calls in the async event loop
We found that blockbuster demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
The ongoing npm phishing campaign escalates as attackers hijack the popular 'is' package, embedding malware in multiple versions.
Security News
A critical flaw in the popular npm form-data package could allow HTTP parameter pollution, affecting millions of projects until patched versions are adopted.
Security News
Bun 1.2.19 introduces isolated installs for smoother monorepo workflows, along with performance boosts, new tooling, and key compatibility fixes.