Asyncio Application Monitor and REPL

0.7.1 · active · verified Mon Apr 13

aiomonitor is a Python library that enhances asyncio applications by providing interactive monitoring and a Python REPL (Read-Eval-Print Loop). It allows developers to inspect the state of running asyncio event loops, tasks, and execute arbitrary asynchronous Python code within the application's context via a telnet interface. Since version 0.6.0, it also includes a web-based UI for task inspection and cancellation. The current version is 0.7.1, with a moderate release cadence.

Warnings

breaking Python 3.8 support was dropped in version 0.7.1. aiomonitor now requires Python 3.9 or higher.
Fix: Upgrade your Python environment to 3.9+ or pin aiomonitor to <0.7.1.
breaking Since version 0.5.0, connecting to the aiomonitor's REPL requires a proper telnet client (e.g., `telnetlib3` or the included `python -m aiomonitor.cli`). Simple socket-to-console redirectors like `nc` will no longer provide advanced features like auto-completion.
Fix: Use `python -m aiomonitor.cli` or a compatible telnet client when connecting to the monitor.
breaking The default port numbers for the terminal UI, web UI, and console access changed in version 0.6.0. The telnet interface moved from 50101 to 20101 and the web UI from 50201 to 20102.
Fix: Update your connection commands to use the new default ports: `telnet localhost 20101` and `http://localhost:20102` for the web UI.
deprecated The `loop` argument for `asyncio.get_event_loop()` is deprecated in Python 3.10+ and should be replaced with `asyncio.get_running_loop()` if an event loop is already running. While `aiomonitor.start_monitor` still accepts a `loop` argument, it's often preferred to rely on `asyncio.get_running_loop()` implicitly within `asyncio.run()` contexts or pass `None` to `start_monitor` if a loop is already set.
Fix: Use `loop = asyncio.get_running_loop()` to explicitly get the current loop, or consider if the `loop` argument to `start_monitor` can be omitted if a running loop is guaranteed.
gotcha Users on Windows might encounter issues with `reuse_port` not being supported by the socket module when using `aiomonitor`, particularly with alternative event loops like `winloop`.
Fix: Check for official recommendations regarding Windows compatibility or consider alternative deployment environments if `reuse_port` conflicts arise.

Install

pip install aiomonitor Install aiomonitor

Imports

aiomonitor
```
import aiomonitor
```
start_monitor
```
from aiomonitor import start_monitor
```

Quickstart

This quickstart demonstrates how to integrate `aiomonitor` into an asyncio application using its context manager interface. It starts a background worker task and keeps the main loop running. Once executed, you can connect via `telnet localhost 20101` or access the web UI at `http://localhost:20102` to inspect and interact with the running application.

import asyncio
import aiomonitor

async def worker():
    print("Worker started...")
    try:
        await asyncio.sleep(100) # Simulate a long-running task
    except asyncio.CancelledError:
        print("Worker cancelled!")
    finally:
        print("Worker finished.")

async def main():
    loop = asyncio.get_running_loop()
    # start_monitor automatically starts telnet (20101) and web (20102) interfaces
    with aiomonitor.start_monitor(loop=loop):
        print("aiomonitor started. Connect via telnet localhost 20101 or browser http://localhost:20102")
        task = loop.create_task(worker())
        # Keep the main loop running indefinitely
        await asyncio.Event().wait()
        # The task might be cancelled from the monitor, so await it for cleanup
        await task

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Application gracefully stopped.")

view raw JSON →