| """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 |
|
|
