Install & Compatibility
Where this runs
tested against v1.2.2 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.920 runs
installs and imports cleanly · install 0.0s · import 0.000s · 17.8MB
glibcpy 3.10–3.920 runs
installs and imports cleanly · install 1.5s · import 0.000s · 18MB
16MB installed
● package 16MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
interrupt
✓ from async_interrupt import interrupt
✗ from async_interrupt import wait_for_async_interrupt
This quickstart demonstrates how to use `wait_for_async_interrupt` to interrupt an asynchronous operation. A `work_task` simulates long-running work. An `interrupt_future` is set after a short delay, which causes the `async with wait_for_async_interrupt` block to raise `asyncio.CancelledError`, effectively interrupting the main flow that is waiting for `work_task`.
import asyncio
from async_interrupt import wait_for_async_interrupt
async def _do_work():
try:
print("Worker: Starting long work...")
await asyncio.sleep(5) # Simulate long-running work
print("Worker: Work finished normally")
except asyncio.CancelledError:
print("Worker: Work was interrupted!")
async def main():
interrupt_future = asyncio.Future()
work_task = asyncio.create_task(_do_work())
# Simulate an external condition that finishes the future after 2 seconds
async def trigger_interrupt_soon():
await asyncio.sleep(2)
print("Main: Signalling interrupt...")
interrupt_future.set_result(None) # Complete the future to trigger interrupt
asyncio.create_task(trigger_interrupt_soon())
print("Main: Waiting for work or interrupt...")
try:
async with wait_for_async_interrupt(interrupt_future):
await work_task # Wait for the work task to finish (or be interrupted)
except asyncio.CancelledError:
print("Main: Caught CancelledError due to future completion, as expected.")
# Give the work_task a moment to process the cancellation
await asyncio.sleep(0.1)
if work_task.done():
print(f"Main: Work task status: done, exception: {work_task.exception()}")
else:
print("Main: Work task is still running (this shouldn't happen if interrupt worked).")
if __name__ == "__main__":
asyncio.run(main())
Debug
Known issues
gotchaThe `wait_for_async_interrupt` context manager raises an `asyncio.CancelledError` in the current task when the watched future completes. Ensure your asynchronous code, particularly `await` expressions within the context, is prepared to handle `CancelledError` gracefully to prevent unexpected behavior or resource leaks.fixWrap `await` calls that might be interrupted in `try...except asyncio.CancelledError` blocks to perform cleanup or use `asyncio.shield` if a specific await should not be cancelled by the interrupt.
affects: All versions
gotchaThe `asyncio.Future` passed to `wait_for_async_interrupt` must be explicitly completed (e.g., `future.set_result(None)` or `future.set_exception(e)`) to trigger the interruption. If the future never reaches a 'done' state, no `CancelledError` will be raised by the context manager.fixAlways ensure a clear mechanism exists to complete the `asyncio.Future` used for interruption, even in error paths, to guarantee the interrupt functionality works as intended.
affects: All versions
gotchaThis library requires Python 3.9 or newer. Attempting to use it on older Python versions will result in `ModuleNotFoundError` during installation or `SyntaxError` at runtime due to modern `asyncio` features.fixUpgrade your Python environment to version 3.9 or newer to ensure compatibility.
affects: <1.0.0 (requires >=3.9)
Upgrade
Version history
1.2.2latest on PyPI · released Feb 22, 2025
Audit
Dependencies
No dependency data recorded yet.