asgi.py

"""ASGI lifespan adapter for the application runtime owner."""

from typing import Any

from loguru import logger
from starlette.types import ASGIApp, Receive, Scope, Send

from .application import ApplicationRuntime, startup_failure_message


class RuntimeASGIApp:
    """Delegate HTTP to FastAPI and lifespan to `ApplicationRuntime`."""

    def __init__(self, app: ASGIApp, runtime: ApplicationRuntime) -> None:
        self.app = app
        self.runtime = runtime

    def __getattr__(self, name: str) -> Any:
        return getattr(self.app, name)

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "lifespan":
            await self.app(scope, receive, send)
            return
        await self._lifespan(receive, send)

    async def _lifespan(self, receive: Receive, send: Send) -> None:
        started = False
        while True:
            message = await receive()
            if message["type"] == "lifespan.startup":
                try:
                    await self.runtime.start()
                except Exception as exc:
                    await send(
                        {
                            "type": "lifespan.startup.failed",
                            "message": startup_failure_message(
                                self.runtime.settings,
                                exc,
                            ),
                        }
                    )
                    return
                started = True
                await send({"type": "lifespan.startup.complete"})
                continue

            if message["type"] == "lifespan.shutdown":
                if started:
                    try:
                        closed = await self.runtime.close()
                    except Exception as exc:
                        logger.error(
                            "Shutdown failed: exc_type={}",
                            type(exc).__name__,
                        )
                        await send({"type": "lifespan.shutdown.failed", "message": ""})
                        return
                    if not closed:
                        await send({"type": "lifespan.shutdown.failed", "message": ""})
                        return
                await send({"type": "lifespan.shutdown.complete"})
                return