162 lines
4.7 KiB
Python
162 lines
4.7 KiB
Python
import asyncio
|
|
from collections.abc import Awaitable, Callable
|
|
|
|
from fastapi import FastAPI
|
|
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
|
|
|
|
from adapter.config.loader import load_config
|
|
from adapter.config.model import AppConfig
|
|
from adapter.di.container import AppContainer, build_container
|
|
from adapter.http.fastapi.dependencies import APP_CONFIG_STATE, APP_CONTAINER_STATE
|
|
from adapter.http.fastapi.middleware import register_middleware
|
|
from adapter.http.fastapi.routers.v1.router import router as v1_router
|
|
|
|
API_V1_PREFIX = '/api/v1'
|
|
APP_CLEANUP_TASK_STATE = 'cleanup_task'
|
|
APP_CLEANUP_STOP_STATE = 'cleanup_stop'
|
|
|
|
|
|
def create_app(config: AppConfig | None = None) -> FastAPI:
|
|
app_config = load_config() if config is None else config
|
|
container = build_container(config=app_config)
|
|
app: FastAPI | None = None
|
|
|
|
try:
|
|
app = FastAPI(title=app_config.app.name)
|
|
setattr(app.state, APP_CONFIG_STATE, app_config)
|
|
setattr(app.state, APP_CONTAINER_STATE, container)
|
|
app.add_event_handler('startup', _build_startup_handler(app, container))
|
|
app.add_event_handler('shutdown', _build_shutdown_handler(app, container))
|
|
register_middleware(app, app_config)
|
|
app.include_router(v1_router, prefix=API_V1_PREFIX)
|
|
|
|
FastAPIInstrumentor.instrument_app(
|
|
app,
|
|
tracer_provider=container.observability.tracer_provider,
|
|
meter_provider=container.observability.meter_provider,
|
|
exclude_spans=['send', 'receive'],
|
|
)
|
|
|
|
return app
|
|
except Exception:
|
|
try:
|
|
if app is not None:
|
|
_uninstrument_app(app)
|
|
finally:
|
|
container.shutdown()
|
|
raise
|
|
|
|
|
|
def _build_startup_handler(
|
|
app: FastAPI,
|
|
container: AppContainer,
|
|
) -> Callable[[], Awaitable[None]]:
|
|
async def startup() -> None:
|
|
task = _get_cleanup_task(app)
|
|
if task is not None and not task.done():
|
|
return
|
|
|
|
await asyncio.to_thread(container.sandbox_reconciler.execute)
|
|
|
|
stop_event = asyncio.Event()
|
|
setattr(app.state, APP_CLEANUP_STOP_STATE, stop_event)
|
|
setattr(
|
|
app.state,
|
|
APP_CLEANUP_TASK_STATE,
|
|
asyncio.create_task(
|
|
_run_cleanup_loop(container, stop_event),
|
|
name='sandbox_cleanup',
|
|
),
|
|
)
|
|
|
|
return startup
|
|
|
|
|
|
def _build_shutdown_handler(
|
|
app: FastAPI,
|
|
container: AppContainer,
|
|
) -> Callable[[], Awaitable[None]]:
|
|
async def shutdown() -> None:
|
|
errors: list[Exception] = []
|
|
|
|
try:
|
|
await _stop_cleanup_loop(app)
|
|
except Exception as exc:
|
|
errors.append(exc)
|
|
|
|
try:
|
|
_uninstrument_app(app)
|
|
except Exception as exc:
|
|
errors.append(exc)
|
|
|
|
try:
|
|
container.shutdown()
|
|
except Exception as exc:
|
|
errors.append(exc)
|
|
|
|
if errors:
|
|
raise ExceptionGroup('app shutdown failed', errors)
|
|
|
|
return shutdown
|
|
|
|
|
|
async def _run_cleanup_loop(
|
|
container: AppContainer,
|
|
stop_event: asyncio.Event,
|
|
) -> None:
|
|
interval = container.config.sandbox.cleanup_interval_seconds
|
|
|
|
while not stop_event.is_set():
|
|
try:
|
|
await asyncio.to_thread(
|
|
container.usecases.cleanup_expired_sandboxes.execute
|
|
)
|
|
except Exception as exc:
|
|
container.observability.logger.error(
|
|
'sandbox_cleanup_failed',
|
|
attrs={
|
|
'error': type(exc).__name__,
|
|
},
|
|
)
|
|
|
|
try:
|
|
await asyncio.wait_for(stop_event.wait(), timeout=interval)
|
|
except asyncio.TimeoutError:
|
|
continue
|
|
|
|
|
|
async def _stop_cleanup_loop(app: FastAPI) -> None:
|
|
stop_event = _get_cleanup_stop_event(app)
|
|
if stop_event is not None:
|
|
stop_event.set()
|
|
|
|
task = _get_cleanup_task(app)
|
|
try:
|
|
if task is not None:
|
|
await task
|
|
finally:
|
|
setattr(app.state, APP_CLEANUP_TASK_STATE, None)
|
|
setattr(app.state, APP_CLEANUP_STOP_STATE, None)
|
|
|
|
|
|
def _get_cleanup_task(app: FastAPI) -> asyncio.Task[None] | None:
|
|
task = getattr(app.state, APP_CLEANUP_TASK_STATE, None)
|
|
if isinstance(task, asyncio.Task):
|
|
return task
|
|
return None
|
|
|
|
|
|
def _get_cleanup_stop_event(app: FastAPI) -> asyncio.Event | None:
|
|
stop_event = getattr(app.state, APP_CLEANUP_STOP_STATE, None)
|
|
if isinstance(stop_event, asyncio.Event):
|
|
return stop_event
|
|
return None
|
|
|
|
|
|
def _uninstrument_app(app: FastAPI) -> None:
|
|
if _is_instrumented(app):
|
|
FastAPIInstrumentor.uninstrument_app(app)
|
|
|
|
|
|
def _is_instrumented(app: FastAPI) -> bool:
|
|
return bool(getattr(app, '_is_instrumented_by_opentelemetry', False))
|