{"id":7911,"library":"aiorun","title":"Aiorun","description":"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.","status":"active","version":"2025.1.1","language":"en","source_language":"en","source_url":"https://github.com/cjrh/aiorun","tags":["asyncio","boilerplate","event loop","graceful shutdown","signals"],"install":[{"cmd":"pip install aiorun","lang":"bash","label":"Install stable version"}],"dependencies":[],"imports":[{"symbol":"run","correct":"from aiorun import run"}],"quickstart":{"code":"import asyncio\nfrom aiorun import run\n\nasync def my_long_running_task(name, duration):\n    try:\n        print(f\"Task {name}: Starting for {duration} seconds...\")\n        for i in range(duration):\n            await asyncio.sleep(1)\n            print(f\"Task {name}: {i + 1}/{duration} seconds passed.\")\n        print(f\"Task {name}: Completed normally.\")\n    except asyncio.CancelledError:\n        print(f\"Task {name}: Was cancelled, performing cleanup...\")\n    finally:\n        print(f\"Task {name}: Exiting.\")\n\nasync def main():\n    print(\"Main application starting...\")\n    # Schedule multiple tasks\n    asyncio.create_task(my_long_running_task(\"Alpha\", 5))\n    asyncio.create_task(my_long_running_task(\"Beta\", 10))\n    \n    # Keep main running until a shutdown signal is received\n    # In aiorun, you don't typically need loop.run_forever() directly\n    # The 'run()' function itself manages the loop until a signal.\n    print(\"Main application tasks scheduled. Waiting for shutdown signal (e.g., Ctrl+C)...\")\n\nif __name__ == '__main__':\n    # aiorun.run() starts the event loop and handles graceful shutdown\n    # It expects an awaitable (coroutine or task) to start the application logic.\n    try:\n        run(main())\n    except KeyboardInterrupt:\n        print(\"\\nApplication terminated by KeyboardInterrupt (Ctrl+C).\")\n    print(\"Application shutdown complete.\")","lang":"python","description":"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."},"warnings":[{"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.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"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.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"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.","message":"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.","severity":"breaking","affected_versions":"All versions"},{"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.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"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.","message":"`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.","severity":"gotcha","affected_versions":"All versions"}],"env_vars":null,"last_verified":"2026-04-16T00:00:00.000Z","next_check":"2026-07-15T00:00:00.000Z","problems":[{"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.","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.","error":"RuntimeError: Event loop is closed"},{"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.","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.","error":"Task exception was never retrieved"},{"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()`.","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.","error":"My tasks don't stop when I press Ctrl+C (or send SIGINT)"}]}