Examples of aiomonitor usage

Below is a list of examples from aiomonitor/examples

Every example is a correct tiny python program.

Basic Usage

Basic example, starts monitor around loop.run_forever() function:

import asyncio

import aiomonitor


async def main() -> None:
    loop = asyncio.get_running_loop()
    with aiomonitor.start_monitor(loop=loop) as monitor:
        print("Now you can connect with:")
        print(f"  python -m aiomonitor.cli -H {monitor.host} -p {monitor.port}")
        print("or")
        print(f"  telnet {monitor.host} {monitor.port}  # Linux only")
        loop.run_forever()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

aiohttp Example

Full feature example with aiohttp application:

import asyncio
import logging

import uvloop
from aiohttp import web

import aiomonitor


async def inner2() -> None:
    await asyncio.sleep(100)


async def inner1(tg: asyncio.TaskGroup) -> None:
    t = tg.create_task(inner2())
    await t


async def simple(request: web.Request) -> web.Response:
    print("Start sleeping")
    async with asyncio.TaskGroup() as tg:
        tg.create_task(inner1(tg))
    print("Finished sleeping")
    return web.Response(text="Simple answer")


async def main() -> None:
    loop = asyncio.get_running_loop()
    app = web.Application()
    app.router.add_get("/simple", simple)
    with aiomonitor.start_monitor(loop, hook_task_factory=True):
        await web._run_app(app, port=8090, host="localhost")


if __name__ == "__main__":
    logging.basicConfig()
    logging.getLogger("aiomonitor").setLevel(logging.DEBUG)
    uvloop.install()
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

Any above examples compatible with uvloop, in fact aiomonitor test suite executed against asyncio and uvloop.

aiohttp Example with additional command

Same as above, but showing a custom command:

import asyncio
from typing import Optional

import click
import requests
import uvloop
from aiohttp import web

import aiomonitor
from aiomonitor.termui.commands import (
    auto_command_done,
    custom_help_option,
    monitor_cli,
)


async def simple(request: web.Request) -> web.Response:
    await asyncio.sleep(10)
    return web.Response(text="Simple answer")


async def hello(request: web.Request) -> web.StreamResponse:
    resp = web.StreamResponse()
    name = request.match_info.get("name", "Anonymous")
    answer = ("Hello, " + name).encode("utf8")
    resp.content_length = len(answer)
    resp.content_type = "text/plain"
    await resp.prepare(request)
    await asyncio.sleep(10)
    await resp.write(answer)
    await resp.write_eof()
    return resp


@monitor_cli.command(name="hello")
@click.argument("name", optional=True)
@custom_help_option
@auto_command_done
def do_hello(ctx: click.Context, name: Optional[str] = None) -> None:
    """Using the /hello GET interface

    There is one optional argument, "name".  This name argument must be
    provided with proper URL excape codes, like %20 for spaces.
    """
    name = "" if name is None else "/" + name
    r = requests.get("http://localhost:8090/hello" + name)
    click.echo(r.text + "\n")


async def main() -> None:
    loop = asyncio.get_running_loop()
    app = web.Application()
    app.router.add_get("/simple", simple)
    app.router.add_get("/hello/{name}", hello)
    app.router.add_get("/hello", hello)
    host, port = "localhost", 8090
    with aiomonitor.start_monitor(loop, locals=locals()):
        web.run_app(app, port=port, host=host)


if __name__ == "__main__":
    uvloop.install()
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass