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 escape 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