{"id":6523,"library":"async-interrupt","title":"Async Interrupt","description":"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.","status":"active","version":"1.2.2","language":"en","source_language":"en","source_url":"https://github.com/bluetooth-devices/async_interrupt","tags":["asyncio","context-manager","concurrency","interruption","future"],"install":[{"cmd":"pip install async-interrupt","lang":"bash","label":"Install stable version"}],"dependencies":[],"imports":[{"symbol":"wait_for_async_interrupt","correct":"from async_interrupt import wait_for_async_interrupt"}],"quickstart":{"code":"import asyncio\nfrom async_interrupt import wait_for_async_interrupt\n\nasync def _do_work():\n    try:\n        print(\"Worker: Starting long work...\")\n        await asyncio.sleep(5)  # Simulate long-running work\n        print(\"Worker: Work finished normally\")\n    except asyncio.CancelledError:\n        print(\"Worker: Work was interrupted!\")\n\nasync def main():\n    interrupt_future = asyncio.Future()\n    \n    work_task = asyncio.create_task(_do_work())\n\n    # Simulate an external condition that finishes the future after 2 seconds\n    async def trigger_interrupt_soon():\n        await asyncio.sleep(2)\n        print(\"Main: Signalling interrupt...\")\n        interrupt_future.set_result(None) # Complete the future to trigger interrupt\n\n    asyncio.create_task(trigger_interrupt_soon())\n\n    print(\"Main: Waiting for work or interrupt...\")\n    try:\n        async with wait_for_async_interrupt(interrupt_future):\n            await work_task # Wait for the work task to finish (or be interrupted)\n    except asyncio.CancelledError:\n        print(\"Main: Caught CancelledError due to future completion, as expected.\")\n    \n    # Give the work_task a moment to process the cancellation\n    await asyncio.sleep(0.1)\n    \n    if work_task.done():\n        print(f\"Main: Work task status: done, exception: {work_task.exception()}\")\n    else:\n        print(\"Main: Work task is still running (this shouldn't happen if interrupt worked).\")\n\nif __name__ == \"__main__\":\n    asyncio.run(main())\n","lang":"python","description":"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`."},"warnings":[{"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.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"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.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"fix":"Upgrade your Python environment to version 3.9 or newer to ensure compatibility.","message":"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.","severity":"gotcha","affected_versions":"<1.0.0 (requires >=3.9)"}],"env_vars":null,"last_verified":"2026-04-15T00:00:00.000Z","next_check":"2026-07-14T00:00:00.000Z","problems":[{"fix":"Install the package using 'pip install async-interrupt'.","cause":"The 'async_interrupt' package is not installed in the Python environment.","error":"ModuleNotFoundError: No module named 'async_interrupt'"},{"fix":"Use 'from async_interrupt import interrupt' to import the function correctly.","cause":"The 'interrupt' function is not directly available in the 'async_interrupt' module.","error":"ImportError: cannot import name 'interrupt' from 'async_interrupt'"},{"fix":"Ensure 'await' is used within an 'async def' function.","cause":"Using 'await' outside of an asynchronous function.","error":"SyntaxError: invalid syntax"},{"fix":"Use 'await' instead of 'asyncio.run()' if the event loop is already running.","cause":"Attempting to run an asyncio event loop that is already running.","error":"RuntimeError: This event loop is already running"},{"fix":"Ensure the function being awaited is asynchronous and returns an awaitable object.","cause":"Awaiting a function that does not return an awaitable object.","error":"TypeError: object NoneType can't be used in 'await' expression"}]}