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.