Handle several database outage situations gracefully, add database health check endpoint

release-notes.md

@@ -7,6 +7,8 @@
 * Add a robots.txt file and allow indexing only of documentation
 * Change mutable list defaults in Pydantic to default factory lists
 * Add tests for empty entregas lists
+* Handle several database outage situations gracefully, add database
+  health check endpoint
 
 
 ## 3.2.5

src/api.py

@@ -4,19 +4,26 @@
 from contextlib import asynccontextmanager
 from datetime import timedelta
 import json
+import logging
 import os
 from textwrap import dedent
 from typing import Annotated, Awaitable, Callable, Union
 
 from fastapi import Depends, FastAPI, HTTPException, status, Header, Request, Response
 from fastapi.security import OAuth2PasswordRequestForm
 from fastapi.responses import JSONResponse, RedirectResponse
-from sqlalchemy.exc import IntegrityError
+from sqlalchemy.exc import IntegrityError, OperationalError
+from sqlalchemy.ext.asyncio import AsyncSession
 
 
 import crud
 import crud_auth
-from db_config import DbContextManager, create_db_and_tables
+from db_config import (
+    check_db_connection,
+    create_db_and_tables,
+    DbContextManager,
+    get_db,
+)
 import email_config
 import response_schemas
 import schemas
@@ -47,12 +54,22 @@
 
     description = environment_msg + description
 
+# Configura o logger
+
+logger = logging.getLogger("uvicorn.error")
+
+# Inicialização da API
+
 
 @asynccontextmanager
-async def lifespan(app: FastAPI):
+async def lifespan(application: FastAPI):  # pylint: disable=unused-argument
     """Executa as rotinas de inicialização da API."""
-    await create_db_and_tables()
-    await crud_auth.init_user_admin()
+    try:
+        await create_db_and_tables()
+        await crud_auth.init_user_admin()
+    except OperationalError as exception:
+        logger.error("A inicialização do banco de dados falhou: %s", exception)
+        raise exception
     yield
 
 
@@ -64,6 +81,9 @@ async def lifespan(app: FastAPI):
 )
 
 
+# Middleware
+
+
 @app.middleware("http")
 async def add_csp_header(
     request: Request, call_next: Callable[[Request], Awaitable[Response]]
@@ -115,6 +135,35 @@ async def check_user_agent(
     return await call_next(request)
 
 
+# Tratadores de exceções
+
+
+@app.exception_handler(OperationalError)
+async def db_exception_handler(
+    request: Request, exception: OperationalError
+) -> JSONResponse:
+    """Trata a exceção de erro operacional do banco de dados, retornando
+    erro 503 Service Unavailable para qualquer endpoint.
+
+    Args:
+        request (Request): Requisição HTTP.
+        exception (OperationalError): Erro operacional do banco de dados.
+
+    Returns:
+        Response: Resposta HTTP informando o erro.
+    """
+    logger.error("Erro operacional do banco de dados em %s: %s", request.url, exception)
+    return JSONResponse(
+        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+        content={
+            "detail": "Banco de dados indisponível. Por favor tente novamente mais tarde."
+        },
+    )
+
+
+# Endpoints
+
+
 @app.get("/", include_in_schema=False)
 async def docs_redirect(
     accept: Union[str, None] = Header(default="text/html")
@@ -131,6 +180,21 @@ async def docs_redirect(
     )
 
 
+@app.get("/health", include_in_schema=False)
+async def health_check(db: AsyncSession = Depends(get_db)):
+    """Verifica se o banco de dados está disponível."""
+    try:
+        # Executa uma query leve para verificar a conectividade com
+        # o banco de dados
+        await check_db_connection(db)
+    except OperationalError as exception:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Banco de dados indisponível",
+        ) from exception
+    return {"status": "ok"}
+
+
 @app.get("/robots.txt", include_in_schema=False)
 async def robots_txt() -> Response:
     """Retorna um arquivo robots.txt para orientar crawlers e permitir

src/db_config.py

@@ -5,10 +5,11 @@
 from typing import AsyncGenerator
 
 from sqlalchemy import create_engine
-from sqlalchemy.orm import DeclarativeBase, sessionmaker
+from sqlalchemy.orm import DeclarativeBase
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.sql import text
 
-SQLALCHEMY_DATABASE_URL = os.environ.get("SQLALCHEMY_DATABASE_URL")
+SQLALCHEMY_DATABASE_URL = os.environ["SQLALCHEMY_DATABASE_URL"]
 
 engine = create_async_engine(SQLALCHEMY_DATABASE_URL)
 sync_engine = create_engine(SQLALCHEMY_DATABASE_URL)
@@ -17,29 +18,59 @@
 
 
 # database models (SQLAlchemy)
-class Base(DeclarativeBase):
-    pass
+class Base(DeclarativeBase):  # pylint: disable=too-few-public-methods
+    """Classe base para modelos SQL Alchemy.
+    """
 
 
 async def create_db_and_tables():
+    """"Inicializa o banco de dados e as tabelas, se não existirem.
+    """
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
 
 
 async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
+    """Retorna a sessão do banco de dados.
+
+    Returns:
+        AsyncGenerator[AsyncSession, None]: gerador assíncrono para
+            sessões do banco de dados.
+
+    Yields:
+        Iterator[AsyncGenerator[AsyncSession, None]]: iterador para
+            receber uma sessão.
+    """
     async with async_session_maker() as session:
         yield session
 
 
-async def get_db():
-    db = get_async_session()
-    try:
+async def get_db() -> AsyncGenerator[AsyncSession, None]:
+    """Retorna sessões do banco de dados.
+
+    Returns:
+    AsyncGenerator[AsyncSession, None]: gerador assíncrono para
+        sessões do banco de dados.
+
+    Yields:
+        Iterator[AsyncGenerator[AsyncSession, None]]: iterador para
+        receber uma sessão.
+    """
+    async for db in get_async_session():
         yield db
-    finally:
-        db.aclose()
+
+
+async def check_db_connection(db: AsyncSession):
+    """Verifica a conectividade com o banco de dados usando uma query
+    leve.
+    """
+    result = await db.execute(text("SELECT 1"))
+    _ = result.scalar_one_or_none()
 
 
 class DbContextManager:
+    """Context manager para manipulação de sessões do banco de dados.
+    """
     def __init__(self):
         self.db = async_session_maker()
 

src/models.py

@@ -10,7 +10,6 @@
     Column,
     Date,
     DateTime,
-    Enum,
     ForeignKeyConstraint,
     Integer,
     String,