Async Interrupt

1.2.2 · active · verified Wed Apr 15

async-interrupt is a Python library providing an `asyncio` context manager that raises an exception in the current task when a specified `asyncio.Future` becomes done. This allows for interrupting long-running or blocking async operations based on external conditions, effectively 'waiting for' a future to complete by interrupting the current task. The current version is 1.2.2, and it follows an infrequent, feature-driven release cadence.

Common errors

```
ModuleNotFoundError: No module named 'async_interrupt'
```
cause The 'async_interrupt' package is not installed in the Python environment.

fix Install the package using 'pip install async-interrupt'.
```
ImportError: cannot import name 'interrupt' from 'async_interrupt'
```
cause The 'interrupt' function is not directly available in the 'async_interrupt' module.

fix Use 'from async_interrupt import interrupt' to import the function correctly.
```
SyntaxError: invalid syntax
```
cause Using 'await' outside of an asynchronous function.

fix Ensure 'await' is used within an 'async def' function.
```
RuntimeError: This event loop is already running
```
cause Attempting to run an asyncio event loop that is already running.

fix Use 'await' instead of 'asyncio.run()' if the event loop is already running.
```
TypeError: object NoneType can't be used in 'await' expression
```
cause Awaiting a function that does not return an awaitable object.

fix Ensure the function being awaited is asynchronous and returns an awaitable object.

Warnings

gotcha The `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.
Fix: Wrap `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.
gotcha The `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.
Fix: Always 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.
gotcha This 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.
Fix: Upgrade your Python environment to version 3.9 or newer to ensure compatibility.

Install

pip install async-interrupt Install stable version

Imports

wait_for_async_interrupt

from async_interrupt import wait_for_async_interrupt

Quickstart

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())

view raw JSON →