Lifespan

Version: 1.1.0 Status: SOURCE OF TRUTH Last Updated: 2026-06-28

Overview

ServerLifespan manages startup and shutdown sequences for AsgiServer.

Responsibilities:

Call sub-app on_startup hooks (in mount order)
Call sub-app on_shutdown hooks (in reverse mount order)
Support sync or async hooks transparently
Report startup failures to the ASGI server via lifespan.startup.failed

Lifespan is exported as an alias of ServerLifespan.

Note: the lifespan does NOT initialize config/logger/executors. Server resources (config handler, session store, storage, resource loader, logger, dispatcher, etc.) are created in AsgiServer.__init__, not during the ASGI lifespan startup event. The lifespan only coordinates the mounted apps’ hooks.

ServerLifespan Class

class ServerLifespan:
    """ASGI Lifespan handler for AsgiServer."""

    __slots__ = ("server", "_logger", "_started")

    def __init__(self, server: AsgiServer) -> None:
        self.server = server
        self._logger = logging.getLogger("genro_asgi.lifespan")
        self._started = False

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """Handle ASGI lifespan events."""
        while True:
            message = await receive()
            msg_type = message["type"]

            if msg_type == "lifespan.startup":
                try:
                    await self.startup()
                    await send({"type": "lifespan.startup.complete"})
                except Exception as e:
                    self._logger.exception("Startup failed")
                    await send({
                        "type": "lifespan.startup.failed",
                        "message": str(e),
                    })
                    return

            elif msg_type == "lifespan.shutdown":
                try:
                    await self.shutdown()
                except Exception:
                    self._logger.exception("Shutdown error")
                finally:
                    await send({"type": "lifespan.shutdown.complete"})
                return

Startup Sequence

Startup iterates the mounted apps in mount order and invokes their on_startup hook if present:

async def startup(self) -> None:
    """Execute startup sequence. Calls on_startup on all mounted apps."""
    self._logger.info("AsgiServer starting up...")

    for path, app in self.server.apps.items():
        if hasattr(app, "on_startup"):
            self._logger.debug(f"Starting app at {path}")
            await self._call_handler(app, "on_startup")

    self._started = True
    self._logger.info("AsgiServer started")

self.server.apps is a dict[str, AsgiApplication] mapping mount name (URL prefix) to the app instance directly. The lifespan iterates the instances; it does not unwrap a handler dict.

Shutdown Sequence

Shutdown iterates the mounted apps in REVERSE mount order. Errors raised by an app’s on_shutdown are logged and swallowed so the remaining apps still shut down:

async def shutdown(self) -> None:
    """Execute shutdown sequence. Calls on_shutdown in reverse order."""
    self._logger.info("AsgiServer shutting down...")

    for path, app in reversed(list(self.server.apps.items())):
        if hasattr(app, "on_shutdown"):
            self._logger.debug(f"Stopping app at {path}")
            try:
                await self._call_handler(app, "on_shutdown")
            except Exception:
                self._logger.exception(f"Error shutting down app at {path}")

    self._started = False
    self._logger.info("AsgiServer stopped")

The lifespan does not call any server-level teardown method; it only runs the apps’ hooks.

Sync or Async Hooks

on_startup / on_shutdown may be either sync or async. The handler is invoked through _call_handler, which awaits the result only if it is awaitable:

async def _call_handler(self, app: object, method_name: str) -> None:
    """Call a handler method on an app (sync or async)."""
    handler = getattr(app, method_name)
    if callable(handler):
        result = handler()
        if hasattr(result, "__await__"):
            await result

Startup/Shutdown Order

Phase	Startup Order	Shutdown Order
Apps	Mount order (`apps.items()`)	Reverse mount order (`reversed(...)`)

Principle: Sub-apps are shut down in reverse order of mounting, so apps mounted later (which may depend on earlier ones) tear down first.

The _server application (system endpoints) is mounted first in AsgiServer.__init__, so it starts before user apps and shuts down last.

Sub-App Hooks

Apps can define on_startup / on_shutdown (sync or async). They are optional: the lifespan checks for them with hasattr, so apps without hooks are skipped.

class MyApp(AsgiApplication):
    async def on_startup(self) -> None:
        """Called when the server starts."""
        self.db = await create_db_pool()
        self.server.logger.info("Database pool created")

    async def on_shutdown(self) -> None:
        """Called when the server stops."""
        await self.db.close()
        self.server.logger.info("Database pool closed")

A mounted app reaches server resources (logger, storage, session store, db_registry) through the dual parent-child relationship as self.server, set by attach_instance on mount. The base AsgiApplication does not define on_startup / on_shutdown; apps add them only when they need lifecycle hooks.

Error Handling

Startup Failure

If startup() raises, the lifespan logs the exception, reports failure to the ASGI server, and stops (it does not wait for a shutdown event):

try:
    await self.startup()
    await send({"type": "lifespan.startup.complete"})
except Exception as e:
    self._logger.exception("Startup failed")
    await send({
        "type": "lifespan.startup.failed",
        "message": str(e),
    })
    return

Shutdown Errors

Per-app shutdown errors are caught inside the loop and logged, so one failing app does not stop the others. A failure anywhere in shutdown() is also caught by the outer try/except in __call__, which still sends lifespan.shutdown.complete in the finally block:

for path, app in reversed(list(self.server.apps.items())):
    if hasattr(app, "on_shutdown"):
        try:
            await self._call_handler(app, "on_shutdown")
        except Exception:
            self._logger.exception(f"Error shutting down app at {path}")
            # Continue with the other apps

ASGI Lifespan Protocol

Lifespan events follow the ASGI spec:

Server                          Uvicorn
------                          -------
                  <--           lifespan.startup
startup sequence
lifespan.startup.complete -->
                                (server running)
                  <--           lifespan.shutdown
shutdown sequence
lifespan.shutdown.complete -->

Usage with AsgiServer

ServerLifespan is created automatically in AsgiServer.__init__ and exposed as server.lifespan. The server’s __call__ routes lifespan-scoped events to it (websocket scopes go to the WsxHandler, everything else to the Dispatcher):

class AsgiServer(BasicAuthMixin, RoutingClass):
    def __init__(self, ...):
        ...
        self.lifespan = ServerLifespan(self)
        ...

    async def __call__(self, scope, receive, send):
        if scope["type"] == "lifespan":
            await self.lifespan(scope, receive, send)
        elif scope["type"] == "websocket":
            await self.wsx_handler(scope, receive, send)
        else:
            await self.dispatcher(scope, receive, send)

Testing Without Lifespan

For testing you can skip the ASGI lifespan protocol and drive the server directly, optionally invoking the startup/shutdown coroutines yourself:

# Direct call without the lifespan protocol
server = AsgiServer()
server.mount("api", api_app)

# Call directly
await server(scope, receive, send)

# Or manually trigger startup/shutdown
await server.lifespan.startup()
try:
    await server(scope, receive, send)
finally:
    await server.lifespan.shutdown()

startup() and shutdown() are public methods on ServerLifespan.