""" FastAPI application factory. Creates and configures the FastAPI app instance with: - Lifespan context (startup/shutdown hooks) - CORS middleware - Global exception handlers (AppException → structured JSON) - API router registration (v1) - Health check endpoint - Prometheus metrics (if enabled) - Sentry error tracking (if configured) """ from collections.abc import AsyncGenerator from contextlib import asynccontextmanager import structlog from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from slowapi.errors import RateLimitExceeded from slowapi.middleware import SlowAPIMiddleware from starlette.middleware.trustedhost import TrustedHostMiddleware from app.config import settings from app.core.exceptions import AppException from app.core.http_middleware import HTTPPolicyMiddleware from app.core.rate_limit import close_rate_limit_storage, limiter logger = structlog.get_logger(__name__) # ── Lifespan ────────────────────────────────────────────────────────────────── @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: """ Application startup and shutdown hooks. Startup: - Seeds system permissions into the database (idempotent). Shutdown: - Disposes the async engine connection pool. """ # ── Startup ─────────────────────────────────────────────────────────────── logger.info("startup_begin") from app.core.state_store import cleanup_expired_state from app.database import AsyncSessionLocal from app.repositories.permission_repository import PermissionRepository from app.repositories.unit_repository import UnitRepository async with AsyncSessionLocal() as session: await PermissionRepository().seed_permissions(session) await UnitRepository().seed_defaults( session, [ "g/dL", "mg/dL", "mmol/L", "mEq/L", "U/L", "IU/L", "mIU/L", "ng/mL", "pg/mL", "ug/dL", "uIU/mL", "10^3/uL", "10^6/uL", "cells/uL", "%", "fL", "pg", "mm/hr", "sec", "min", "ratio", "Index", "mg/L", "IU/mL", "ng/dL", "/hpf", "Ct", ], ) await session.commit() logger.info("startup_complete") await cleanup_expired_state() yield # Application runs here # ── Shutdown ────────────────────────────────────────────────────────────── logger.info("shutdown_begin") from app.database import engine await engine.dispose() close_rate_limit_storage() logger.info("shutdown_complete") # ── Exception Handlers ──────────────────────────────────────────────────────── async def app_exception_handler(request: Request, exc: AppException) -> JSONResponse: """ Convert any AppException subclass into a typed ErrorResponse JSON payload. This is the single place that translates application errors into HTTP responses. Routers never call JSONResponse directly for error cases — they raise instead. """ logger.warning( "app_exception", code=exc.code, path=getattr(request.scope.get("route"), "path", "unmatched"), method=request.method, ) return JSONResponse( status_code=exc.status_code, content={ "success": False, "error": { "code": exc.code, "message": exc.message, }, }, ) async def rate_limit_exceeded_handler(request: Request, exc: RateLimitExceeded) -> JSONResponse: """Format rate limit violations into the standard Munal LMS error envelope.""" logger.warning( "rate_limit_exceeded", path=getattr(request.scope.get("route"), "path", "unmatched"), detail=str(exc.detail), ) return JSONResponse( status_code=429, content={ "success": False, "error": { "code": "RATE_LIMIT_EXCEEDED", "message": f"Too many requests: {exc.detail}", }, }, ) # ── Application Factory ─────────────────────────────────────────────────────── def create_app() -> FastAPI: """ Application factory pattern. Returns a fully configured FastAPI instance. Import the module-level `app` for uvicorn/gunicorn — do not call this directly. """ _app = FastAPI( title=settings.APP_NAME, version=settings.APP_VERSION, description=( "Munal Laboratory Management System — REST API\n\n" "All responses use a consistent envelope:\n" '- Success: `{ "success": true, "data": ... }`\n' '- Error: `{ "success": false, "error": { "code": "...", "message": "..." } }`' ), lifespan=lifespan, # Swagger UI and ReDoc are only available in non-production environments. docs_url="/docs" if not settings.is_production else None, redoc_url="/redoc" if not settings.is_production else None, openapi_url="/openapi.json" if not settings.is_production else None, ) # ── Rate Limiter State ──────────────────────────────────────────────────── _app.state.limiter = limiter # ── Exception Handlers ───────────────────────────────────────────────────── _app.add_exception_handler(AppException, app_exception_handler) # type: ignore[arg-type] _app.add_exception_handler(RateLimitExceeded, rate_limit_exceeded_handler) # type: ignore[arg-type] # ── Middleware ───────────────────────────────────────────────────────────── if settings.APP_ALLOWED_HOSTS: _app.add_middleware(TrustedHostMiddleware, allowed_hosts=settings.APP_ALLOWED_HOSTS) _app.add_middleware(HTTPPolicyMiddleware) _app.add_middleware(SlowAPIMiddleware) _app.add_middleware( CORSMiddleware, allow_origins=settings.APP_ALLOWED_ORIGINS, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ── Sentry ──────────────────────────────────────────────────────────────── if settings.SENTRY_DSN: import sentry_sdk from sentry_sdk.integrations.fastapi import FastApiIntegration from sentry_sdk.integrations.starlette import StarletteIntegration sentry_sdk.init( dsn=settings.SENTRY_DSN, environment=settings.APP_ENV, integrations=[StarletteIntegration(), FastApiIntegration()], traces_sample_rate=0.1, ) logger.info("sentry_initialized", dsn_configured=True) # ── Prometheus Metrics ──────────────────────────────────────────────────── if settings.PROMETHEUS_ENABLED: from prometheus_fastapi_instrumentator import Instrumentator Instrumentator().instrument(_app).expose(_app, endpoint="/metrics") # ── API Routers ─────────────────────────────────────────────────────────── from app.api.router import api_router _app.include_router(api_router, prefix="/api") # ── Health & Readiness Probes ───────────────────────────────────────────── @_app.get("/health", tags=["Health"], summary="Liveness probe") async def health_check() -> dict[str, str]: """Returns API liveness status and version. Used by load balancers and monitoring.""" return {"status": "ok", "version": settings.APP_VERSION, "env": settings.APP_ENV} @_app.get("/health/ready", tags=["Health"], summary="Readiness probe") async def readiness_check() -> JSONResponse: """Verify database connectivity and the migrated session-state table.""" from sqlalchemy import text from app.database import AsyncSessionLocal db_ok = False state_ok = False try: async with AsyncSessionLocal() as session: await session.execute(text("SELECT 1")) db_ok = True await session.execute(text("SELECT 1 FROM expiring_state LIMIT 1")) state_ok = True except Exception as exc: logger.error("readiness_check_db_failed", error=str(exc)) status_code = 200 if (db_ok and state_ok) else 503 return JSONResponse( status_code=status_code, content={ "status": "ready" if (db_ok and state_ok) else "degraded", "database": "connected" if db_ok else "unreachable", "session_store": "ready" if state_ok else "unavailable", "version": settings.APP_VERSION, "env": settings.APP_ENV, }, ) logger.info( "app_created", name=settings.APP_NAME, version=settings.APP_VERSION, env=settings.APP_ENV, ) return _app # Module-level app instance — used by: uvicorn app.main:app app = create_app()