feat: add health check endpoints and database health check functionality

Author: umbertogriffo

src/api/endpoints/health.py

@@ -1,3 +1,4 @@
+from database import check_health_safe, engine
 from fastapi import APIRouter
 from starlette import status
 
@@ -14,4 +15,50 @@ def health():
     """
     Performs a simple health check.
     """
-    return {"status": "OK"}
+    return health_check(include_dependencies=False)
+
+
+@router.get(
+    "/health/readiness",
+    operation_id="isReady",
+    summary="Check if the service and dependencies are healthy; i.e. capable of processing requests.",
+    status_code=status.HTTP_200_OK,
+    response_description="Return HTTP 200",
+)
+async def readiness_check():
+    return health_check(include_dependencies=True)
+
+
+@router.get(
+    "/health/liveness",  # Explicit requirement
+    operation_id="isAlive",
+    summary="Check if the service is healthy; i.e. capable of responding to requests.",
+    status_code=status.HTTP_200_OK,
+    response_description="Return HTTP 200",
+)
+async def liveness_check():
+    return health_check(include_dependencies=False)
+
+
+def health_check(include_dependencies: bool = False):
+    """
+    Health check endpoint.
+
+    Args:
+        include_dependencies (bool): Whether to include dependency checks.
+
+    Returns:
+        dict: Health status.
+    """
+    # Here you would typically check the health of your dependencies
+    # e.g., database connection, external services, etc.
+    # For simplicity, we return a static response.
+
+    status_response = {"status": "OK"}
+
+    if include_dependencies:
+        # Perform checks for dependencies here
+        db_status = check_health_safe(engine)
+        status_response["database"] = "OK" if db_status else "FAIL"
+
+    return status_response

src/database.py

@@ -1,6 +1,9 @@
 from config import settings
-from sqlalchemy import Engine
+from sqlalchemy import Engine, text
 from sqlmodel import SQLModel, create_engine
+from utils.log import get_logger
+
+logger = get_logger()
 
 
 def create_db_engine(verbose: bool = False, **kwargs):
@@ -30,6 +33,39 @@ def create_db_engine(verbose: bool = False, **kwargs):
     return engine
 
 
+def check_health(engine: Engine) -> None:
+    """
+    Perform a health check on the database by executing a simple query.
+
+    Args:
+        engine (Engine): The SQLAlchemy engine instance used to connect to the database.
+
+    Raises:
+        Exception: If the database health check fails, the exception is logged and re-raised.
+    """
+    try:
+        with engine.connect() as conn:
+            conn.execute(text("SELECT 1"))
+
+    except Exception as e:
+        logger.error(f"Database health check failed: {e}")
+        raise
+
+
+def check_health_safe(engine: Engine) -> bool:
+    """
+    Safe version of check_health that returns a bool instead of raising for convenience.
+
+    Returns:
+        True if the database is healthy and permissions are valid, False otherwise.
+    """
+    try:
+        check_health(engine)
+        return True
+    except Exception:
+        return False
+
+
 def create_db_and_tables(engine: Engine):
     """
     Create database tables based on the defined SQLModel models.

tests/conftest.py

@@ -17,7 +17,10 @@ def data_folder_path():
 
 @pytest.fixture(name="session")
 def session_fixture() -> Session:
-    """Create an engine with all migrations applied."""
+    """Create a new database session for a test."""
+    # TODO: Use an in-memory SQLite database for faster tests if possible.
+    #       https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/#memory-database
+
     # Create a temporary database file
     fd, path = tempfile.mkstemp(suffix=".db")
     os.close(fd)
@@ -45,6 +48,8 @@ def session_fixture() -> Session:
 
 @pytest.fixture(name="client_with_db")
 def client_fixture(session: Session):
+    """Create a TestClient that uses the test database session."""
+
     def get_session_override():
         return session