Aiorun

2025.1.1 verified Thu Apr 16 auth: no python

Aiorun is a Python library that simplifies the creation of `asyncio` applications by providing a `run()` function to manage common boilerplate for startup and graceful shutdown. It automatically handles event loop creation, task scheduling, and signal handling for `SIGINT` and `SIGTERM` (or CTRL-C/CTRL-BREAK on Windows). Aiorun is actively maintained, with a versioning scheme that often reflects the year of release.

pip install aiorun

Common errors

error RuntimeError: Event loop is closed ↓

cause This often occurs when attempting to interact with the asyncio event loop or schedule tasks after `aiorun.run()` has completed its execution and closed the loop, or if an event loop created independently is being used after `aiorun` has started its own.

fix

Ensure all asyncio operations, including task creation, are initiated from within the primary coroutine passed to aiorun.run() or its descendants. Avoid manually closing the loop or interacting with a loop that aiorun has already managed to completion.

error Task exception was never retrieved ↓

cause An exception occurred in an `asyncio` task, but no `try...except` block caught it within the task, and no custom `loop.set_exception_handler()` was configured to process it. `aiorun` logs these by default but does not terminate.

fix

Add try...except blocks within your async functions to handle expected exceptions gracefully. For unhandled exceptions across all tasks, implement a custom loop.set_exception_handler() to log, alert, or shut down the application as appropriate.

error My tasks don't stop when I press Ctrl+C (or send SIGINT) ↓

cause While `aiorun` installs signal handlers to initiate graceful shutdown, tasks that do not handle `asyncio.CancelledError` or are protected by `aiorun.shutdown_waits_for()` will not stop immediately upon cancellation requests. Also, on Windows, signal handling is different.

fix

Modify your async tasks to include try...except asyncio.CancelledError blocks to perform any necessary cleanup when a cancellation is requested. If a task *must* complete before shutdown, wrap its awaitable in aiorun.shutdown_waits_for().

Warnings

gotcha Unlike `asyncio.run()`, `aiorun.run()` (especially when called without a coroutine or with a coroutine that eventually finishes) will run the event loop indefinitely (`loop.run_forever()`) unless `loop.stop()` is explicitly called or a shutdown signal is received. This can be surprising if you expect the program to exit when the initial coroutine completes. ↓

fix To stop `aiorun` when your main coroutine finishes, ensure `loop.stop()` is called within that coroutine. For typical server applications, `aiorun` is designed to run continuously until an external signal (e.g., `SIGINT`, `SIGTERM`) triggers its graceful shutdown mechanism.

gotcha Unhandled exceptions within `asyncio` tasks do not automatically cause `aiorun` to terminate the program by default. The event loop will continue running, and the exception will be logged but not propagated out of `aiorun.run()`, potentially masking critical failures. ↓

fix Implement a custom exception handler using `loop.set_exception_handler()` if you need specific logic (e.g., program termination, alerting) for unhandled task exceptions. Alternatively, ensure critical tasks are wrapped in `try...except` blocks to handle exceptions gracefully within the task itself.

breaking The standard `asyncio.shield()` function does not protect coroutines from cancellation during `aiorun`'s graceful shutdown sequence because `shield()`'s internal task is also subject to cancellation. This can lead to tasks being unexpectedly interrupted during shutdown. ↓

fix Use `aiorun.shutdown_waits_for()` to wrap coroutines that must complete their execution without interruption during the shutdown process. This function provides similar shielding behavior tailored for `aiorun`'s shutdown mechanism.

gotcha On Windows, `aiorun` cannot intercept `SIGINT` or `SIGTERM` signals directly like on Unix-like systems. Instead, it relies on `CTRL-C` and `CTRL-BREAK` events generated when run in a console window. This limits its signal handling capabilities in certain deployment environments. ↓

fix Be aware of these limitations when deploying `aiorun` applications on Windows. For robust shutdown in non-console Windows environments (e.g., as a service), consider alternative methods for signaling termination to your `aiorun` application, such as monitoring a file or a network port.

gotcha `aiorun.run(coro)` creates a new event loop instance each time it's invoked. This can cause confusing errors if your application code directly interacts with or expects a global 'default' event loop instance provided by the standard library's `asyncio` that is different from the one `aiorun` is managing. ↓

fix It is generally recommended to create and manage all `asyncio` tasks and resources *within* the initial coroutine passed to `aiorun.run()`, or within coroutines spawned from it. This ensures all operations occur on the same event loop managed by `aiorun`, reducing ambiguity. Avoid explicitly calling `asyncio.get_event_loop()` outside this context if possible.

Imports

run
```
from aiorun import run
```

Quickstart

This quickstart demonstrates a basic `aiorun` application with two simulated long-running tasks. It shows how to schedule tasks and includes `asyncio.CancelledError` handling within a task to perform cleanup during a graceful shutdown initiated by a signal (like `Ctrl+C`). The `aiorun.run(main())` call manages the event loop lifecycle and signal processing.

import asyncio
from aiorun import run

async def my_long_running_task(name, duration):
    try:
        print(f"Task {name}: Starting for {duration} seconds...")
        for i in range(duration):
            await asyncio.sleep(1)
            print(f"Task {name}: {i + 1}/{duration} seconds passed.")
        print(f"Task {name}: Completed normally.")
    except asyncio.CancelledError:
        print(f"Task {name}: Was cancelled, performing cleanup...")
    finally:
        print(f"Task {name}: Exiting.")

async def main():
    print("Main application starting...")
    # Schedule multiple tasks
    asyncio.create_task(my_long_running_task("Alpha", 5))
    asyncio.create_task(my_long_running_task("Beta", 10))
    
    # Keep main running until a shutdown signal is received
    # In aiorun, you don't typically need loop.run_forever() directly
    # The 'run()' function itself manages the loop until a signal.
    print("Main application tasks scheduled. Waiting for shutdown signal (e.g., Ctrl+C)...")

if __name__ == '__main__':
    # aiorun.run() starts the event loop and handles graceful shutdown
    # It expects an awaitable (coroutine or task) to start the application logic.
    try:
        run(main())
    except KeyboardInterrupt:
        print("\nApplication terminated by KeyboardInterrupt (Ctrl+C).")
    print("Application shutdown complete.")