lifespan.py

class LifespanManager:
    def __init__(self, app):
        self.app = app
        self.startup_complete = asyncio.Event()
        self.shutdown_complete = asyncio.Event()
        self.messages = [{'type': 'lifespan.startup'}, {'type': 'lifespan.shutdown'}]

    async def __aenter__(self):
        self.task = asyncio.create_task(self.app(self.receive, self.send))
        await self.startup_complete.wait()

    async def __aexit__(self, ...):
        await self.shutdown_complete.wait()
        await self.task
        
    async def receive(self):
        return self.messages.pop()

    async def send(self, message):
        if message['type'] == 'lifespan.startup.complete':
            self.startup_complete.set()
        elif message['type'] == 'lifespan.shutdown.complete':
            self.shutdown_complete.set()

Usage:

async with LifespanManager(app):
    ...  # Do stuff

I like the LifespanManager name. :)

Isn't scope missing on L9?

-        self.task = asyncio.create_task(self.app(self.receive, self.send))
+        scope = {"type": "lifespan"}
+        self.task = asyncio.create_task(self.app(scope, self.receive, self.send))

Yup, good quick catch there.

This is a good asyncio implementation I think (although I'm not sure whether exceptions raised during startup would be handled correctly?) — the implementation in asgi-lifespan follows the same idea with handling of various corner cases and trio/curio support as well.

I really like your approach of popping events out of a list in receive() — going to use it to simplify florimondmanca/asgi-lifespan#2. :)

Yeah in its current state it doesn’t deal with the failed startup or shutdown cases.