Proper way to run code on startup and shutdown

[specd-tech]

I’m trying to run some code before uvicorn starts and after all connections are closed. I have tried to follow the implementation of Starlette’s events but I can’t figure out how they work. The best implementation I can come up with is monkey patching my code to uvicorns startup and shutdown functions.

A simple example of what I’m trying to do:

Before serving requests run this
client = httpx.AsyncClient()

After finishing all requests run this
await client.aclose()

[euri10]

what you want is described as events in Starlette: https://www.starlette.io/events/

[specd-tech]

I’ve read that but I’m trying to implement something that just uses Uvicorn. I have tried to follow Starlette’s source code to see how it implemented lifespan handling. I found out about ASGI lifespans since the original post but I’ve only found this example of how to use them.

https://asgi.readthedocs.io/en/latest/specs/lifespan.html

Is this the best way to use lifespans?

If it is, do I just call Uvicorn on that app function?

[euri10]

if you want to just use uvicorn you'll have to send your own lifespan events, a good intro could be reading the test_lifespan for instance.

other than that, you maybe interested in using the excellent https://github.com/florimondmanca/asgi-lifespan !

[specd-tech]

I found a working solution, I believe this is the correct way to use lifespan events. If anyone has any improvements they would be greatly appreciated.

async def app(scope, receive, send):
    if scope["type"] == "http":
        await send(
            {
                "type": "http.response.start",
                "status": 200,
                "headers": [
                    [b"content-type", b"text/plain"],
                ],
            }
        )
        await send(
            {
                "type": "http.response.body",
                "body": b"Hello, world!",
            }
        )

    elif scope["type"] == "lifespan":
        while True:
            message = await receive()
            if message["type"] == "lifespan.startup":
                try:
                    # startup code goes here
                    await send({"type": "lifespan.startup.complete"})
                except Exception:
                    await send({"type": "lifespan.startup.failed"})
            elif message["type"] == "lifespan.shutdown":
                try:
                    # shutdown code goes here
                    await send({"type": "lifespan.shutdown.complete"})
                except Exception:
                    await send({"type": "lifespan.shutdown.failed"})
                return

[euri10]

this should work yes,
I'm just curious, any reason you want to write pure a asgi app, I mean frameworks provide you some good abstractions, the lifespan handling being one ?

[specd-tech]

Mainly as a challenge and to learn more about ASGI. Also for any possible performance gains. I’m attempting to write a load balancer and wanted to keep my application’s processing to a minimum and handle everything on the actual web servers.