Class LifespanManager

gunicorn/asgi/lifespan.py:16–178

Manages ASGI lifespan events (startup/shutdown). The lifespan protocol allows ASGI applications to run code at startup and shutdown. This is essential for applications that need to initialize database connections, caches, or other resources. ASGI lifespan messages: - Server

class LifespanManager:
    """Manages ASGI lifespan events (startup/shutdown).

    The lifespan protocol allows ASGI applications to run code at
    startup and shutdown. This is essential for applications that
    need to initialize database connections, caches, or other
    resources.

    ASGI lifespan messages:
    - Server sends: {"type": "lifespan.startup"}
    - App responds: {"type": "lifespan.startup.complete"} or
                    {"type": "lifespan.startup.failed", "message": "..."}
    - Server sends: {"type": "lifespan.shutdown"}
    - App responds: {"type": "lifespan.shutdown.complete"}
    """

    def __init__(self, app, logger, state=None):
        """Initialize the lifespan manager.

        Args:
            app: ASGI application callable
            logger: Logger instance
            state: Shared state dict for the application
        """
        self.app = app
        self.logger = logger
        self.state = state if state is not None else {}

        self._startup_complete = asyncio.Event()
        self._shutdown_complete = asyncio.Event()
        self._startup_failed = False
        self._startup_error = None
        self._shutdown_error = None
        self._receive_queue = asyncio.Queue()
        self._task = None
        self._app_finished = False

    async def startup(self):
        """Run lifespan startup and wait for completion.

        Raises:
            RuntimeError: If startup fails or app doesn't support lifespan
        """
        scope = {
            "type": "lifespan",
            "asgi": {"version": "3.0", "spec_version": "2.4"},
            "state": self.state,
        }

        # Send startup event
        await self._receive_queue.put({"type": "lifespan.startup"})

        # Run lifespan in background task
        self._task = asyncio.create_task(self._run_lifespan(scope))

        # Wait for startup with timeout
        try:
            await asyncio.wait_for(