aiorun

library2025.1.1pypypi✓ verified 85d ago

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.

http-networking web-framework workflow

pip install aiorun

github ↗package ↗

INSTALL

IMPORT

SIG · AIORUN

aiorun

http-networkingpythonv2025.1.1

Install

1.5s avg

Import

200ms

Disk

16MB

Pass rate

10/ 10

Env Coverage10 / 10

glibc

3.9–3.13

musl

3.9–3.13

Install & Compatibility

Where this runs

tested against v2025.1.1 · 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

musl

py 3.10–3.920 runs

installs and imports cleanly · install 0.0s · import 0.214s · 17.8MB

glibc

py 3.10–3.920 runs

installs and imports cleanly · install 1.5s · import 0.186s · 18MB

16MB installed

● package 16MB

Code

Verified usage

Verified import paths — ran on the pinned version, not inferred.

run

✓ from aiorun import run

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.")

Debug

Known issues

gotchaUnlike `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.

affects: All versions

gotchaUnhandled 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.

affects: All versions

breakingThe 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.

affects: All versions

gotchaOn 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.

affects: All versions

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.

affects: All versions

Errors

Common errors & fixes

RuntimeError: Event loop is closed

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.

Task exception was never retrieved

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.

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

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

Upgrade

Version history

2025.1.1latest on PyPI · released Jan 27, 2025

Audit

Dependencies

No dependency data recorded yet.

Agent activity

50 hits · last 30 days

node

OpenAI (training)

Resources

githubhttps://github.com/cjrh/aiorun ↗

packagehttps://pypi.org/project/aiorun/ ↗