master/adapter/http/fastapi/app.py

import asyncio
from collections.abc import Awaitable, Callable
from pathlib import Path

from docker.errors import NotFound
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 _ensure_sandbox_network(container: AppContainer) -> None:
    client = container._docker_client
    network_name = container.config.sandbox.network_name
    try:
        client.networks.get(network_name)
    except NotFound:
        client.networks.create(network_name)
        container.observability.logger.info(
            'sandbox_network_created',
            attrs={'network': network_name},
        )


def _ensure_sandbox_dirs(container: AppContainer) -> None:
    cfg = container.config.sandbox
    for path_str in (cfg.dependencies_host_path, cfg.lambda_tools_host_path):
        Path(path_str).mkdir(parents=True, exist_ok=True)


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(_ensure_sandbox_network, container)
        await asyncio.to_thread(_ensure_sandbox_dirs, container)
        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))