Spaces:
Sleeping
Sleeping
File size: 2,358 Bytes
2415446 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 | """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
|