Aiomonitor Integration with Litestar. Uvicorn. FastApi

As aiomonitor primiraly developed for observing aiohttp internal work, i didn't find an example of wrapping up an ASGI application to monitor async tasks. In my case, an app is built on Litestar framework, which allows running ASGI app with uvicorn as a standalone server process.

Internally, uvicorn sets up an event loop accordingly to its input option loop , which is not a loop itself, but a loop type! That is exactly the issue, as you can't wrap the server running line following the way aiomonitor describes. It doesn't work even if you make a couple of changes.

import asyncio
import uvicorn
import aiomonitor

from app import app

loop = asyncio.get_running_loop()  # Error, loop isn't running here
with aiomonitor.start_monitor(loop, locals=locals(), hook_task_factory=True):
	uvicorn.run(app, loop=loop) # Error, loop option means "loop type"

The good news is that it monitors nothing else, but an asyncio loop. Therefore, the loop should be obtained in a proper context to fix the issue.

Solution

After brief reading the source code of Litestar, specifically the Server class, i figured out that the simpliest and most proper way to let aiomonitor do its job is to get the running loop in the life cycle hooks lifespan or on_startup, on_shutdown.

The loop is guaranteed to be running inside the hooks and the monitor can be started and closed with its corresponding methods instead of a context manager.

The code demonstrates the core idea, you should adopt it to your own case.

import uvicorn
from litestar import Litestar
from litestar.datastructures import State
from aiomonitor import start_monitor
from aiomonitor.monitor import Monitor


async def on_stratup(app: Litestar):
	loop = asyncio.get_running_loop()  # Success, loop is running
	monitor: Monitor = start_monitor(loop, locals=locals(), hook_task_factory=True,
									 host='127.0.0.1', port=20101,
									 console_port=20103, webui_port=20102)
	app.state.monitor = monitor
  

def on_shutdown(app: Litestar):
	app.state.monitor.close() # Manually close as context manager isn't used


app = Litestar(
	..., # Your app configuration
	on_startup=[on_stratup],
	on_shutdown=[on_shutdown],
	state=State(),  # State object to access the monitor in the hooks
)

uvicorn.run(app)

Similarly, monitor can reside in lifespan , and you don't even have to create a transfer object State to assign monitor object to, making a life cycle handling cleaner.

FastAPI Intergation

As FastAPI is very close in its concept to the Litestar framework and uses underlying ASGI library Starlette , the solution is 100% compatible with it.

Outcome

After hooks extended, all the functionality is working fine with no issues. Web UI, Telnet and Console UI of aiomonitor work as expected.