"""FastAPI entrypoint with dependency-free liveness and database readiness probes.""" from typing import Any import psycopg import uvicorn from fastapi import FastAPI, Response, status from app import __version__ from app.api.v1 import demo_router from app.core.config import get_settings from app.core.secrets import SecretFileError app = FastAPI(title="Geological RAG API", version=__version__) app.include_router(demo_router) type HealthPayload = dict[str, str | dict[str, str]] @app.get("/health/live", tags=["health"]) @app.get("/api/v1/health/live", tags=["health"]) def live() -> dict[str, str]: return {"status": "ok", "version": __version__} @app.get( "/health/ready", tags=["health"], responses={status.HTTP_503_SERVICE_UNAVAILABLE: {"description": "Database unavailable"}}, ) @app.get("/api/v1/health/ready", tags=["health"]) def ready(response: Response) -> HealthPayload: settings = get_settings() try: dsn = settings.database_url().set(drivername="postgresql") with psycopg.connect( dsn.render_as_string(hide_password=False), connect_timeout=2, autocommit=True, ) as connection: result = connection.execute("SELECT 1").fetchone() if result != (1,): raise psycopg.DatabaseError("readiness query returned an unexpected result") except (OSError, SecretFileError, psycopg.Error): # Do not expose connection strings, secret paths, hostnames, or driver errors. response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE return {"status": "not_ready", "checks": {"database": "unavailable"}} return {"status": "ready", "checks": {"database": "ok"}} @app.get("/api/v1/meta", tags=["meta"]) def meta() -> dict[str, Any]: settings = get_settings() return { "name": settings.app_name, "environment": settings.app_env, "version": __version__, "models": { "embedding": settings.embedding_model, "rerank": settings.rerank_model, "generation": settings.llm_model, }, } if __name__ == "__main__": # Compose publishes this listener only on the host loopback interface. uvicorn.run("app.main:app", host="0.0.0.0", port=8000) # noqa: S104