from typing import Any, Dict from fastapi import APIRouter, Depends, Query from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm import Session from app.core.orchestration import get_orchestration_status from app.db.session import get_db router = APIRouter() @router.get("") def health_check( db: Session = Depends(get_db), detailed: bool = Query(False, description="Include detailed diagnostics information") ) -> Dict[str, Any]: """ Health check endpoint for the API This endpoint checks the health of various components of the application: - Database connectivity - Host and pod status - Application readiness Args: detailed: Whether to include detailed diagnostics information Returns: Dict[str, Any]: Health status of the application """ # Basic health response response = { "status": "healthy" } # Check database connectivity try: # Simple query to verify database connectivity db.execute("SELECT 1") database_status = "connected" database_error = None except SQLAlchemyError as e: database_status = "error" database_error = str(e) except Exception as e: database_status = "error" database_error = str(e) response["database"] = { "status": database_status } # Add orchestration status if detailed information is requested if detailed: response["orchestration"] = get_orchestration_status() # Add more detailed database information if there was an error if database_error: response["database"]["error"] = database_error elif database_status != "connected": # Always include error information in the basic response response["status"] = "unhealthy" response["database"]["error"] = database_error # Set overall status based on component statuses if database_status != "connected": response["status"] = "unhealthy" return response