plantillas-proyectos/backend/core/error_handlers.py

"""
Manejadores globales de excepciones para FastAPI
"""

import logging
from typing import Any, Dict

from fastapi import Request, status
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from sqlalchemy.exc import IntegrityError, SQLAlchemyError

from .config import settings
from .exceptions import BaseAPIException

logger = logging.getLogger(__name__)


def _cors_headers(request: Request) -> Dict[str, str]:
    """CORS headers for error responses so browser does not block on 4xx/5xx."""
    origin = request.headers.get("origin")
    if not origin or origin not in settings.cors_origins_list:
        return {}
    return {
        "Access-Control-Allow-Origin": origin,
        "Access-Control-Allow-Credentials": "true",
    }


async def base_exception_handler(
    request: Request,
    exc: BaseAPIException,
) -> JSONResponse:
    """
    Manejador para todas las excepciones personalizadas de la API
    """
    logger.warning(
        f"API Exception: {exc.error_code} - {exc.message}",
        extra={
            "path": request.url.path,
            "method": request.method,
            "status_code": exc.status_code,
        },
    )

    # Log detailed errors if they exist
    if hasattr(exc, "errors") and exc.errors:
        logger.warning(f"Validation errors details: {exc.errors}")

    response = JSONResponse(
        status_code=exc.status_code,
        content=jsonable_encoder(exc.to_dict()),
    )
    for k, v in _cors_headers(request).items():
        response.headers[k] = v
    return response


async def validation_exception_handler(
    request: Request,
    exc: RequestValidationError,
) -> JSONResponse:
    """
    Manejador para errores de validación de Pydantic/FastAPI
    """
    errors = []
    for error in exc.errors():
        field = ".".join(str(loc) for loc in error["loc"] if loc != "body")
        errors.append(
            {
                "field": field,
                "message": error["msg"],
                "type": error["type"],
            }
        )

    logger.warning(
        f"Validation Error en {request.url.path}",
        extra={"errors": errors},
    )

    response = JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "error": "VALIDATION_ERROR",
            "message": "Error de validación en los datos recibidos",
            "status_code": status.HTTP_422_UNPROCESSABLE_ENTITY,
            "errors": errors,
        },
    )
    for k, v in _cors_headers(request).items():
        response.headers[k] = v
    return response


async def integrity_error_handler(
    request: Request,
    exc: IntegrityError,
) -> JSONResponse:
    """
    Manejador para errores de integridad de la base de datos
    """
    logger.error(
        f"Database Integrity Error: {str(exc.orig)}",
        extra={
            "path": request.url.path,
            "method": request.method,
        },
    )

    # Intentar extraer información útil del error
    error_message = "Error de integridad en la base de datos"

    orig_msg = str(exc.orig).lower()
    if "unique constraint" in orig_msg or "duplicate key" in orig_msg:
        error_message = "El registro ya existe. Verifica los campos únicos."
    elif "foreign key" in orig_msg:
        error_message = "Referencia inválida a otro registro."
    elif "not null" in orig_msg:
        error_message = "Falta un campo requerido."

    response = JSONResponse(
        status_code=status.HTTP_409_CONFLICT,
        content={
            "error": "DATABASE_INTEGRITY_ERROR",
            "message": error_message,
            "status_code": status.HTTP_409_CONFLICT,
        },
    )
    for k, v in _cors_headers(request).items():
        response.headers[k] = v
    return response


async def sqlalchemy_error_handler(
    request: Request,
    exc: SQLAlchemyError,
) -> JSONResponse:
    """
    Manejador para errores generales de SQLAlchemy
    """
    logger.error(
        f"Database Error: {str(exc)}",
        extra={
            "path": request.url.path,
            "method": request.method,
        },
        exc_info=True,
    )

    response = JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            "error": "DATABASE_ERROR",
            "message": "Error en la operación de base de datos",
            "status_code": status.HTTP_500_INTERNAL_SERVER_ERROR,
        },
    )
    for k, v in _cors_headers(request).items():
        response.headers[k] = v
    return response


async def general_exception_handler(
    request: Request,
    exc: Exception,
) -> JSONResponse:
    """
    Manejador para excepciones no capturadas
    """
    logger.error(
        f"Unhandled Exception: {str(exc)}",
        extra={
            "path": request.url.path,
            "method": request.method,
        },
        exc_info=True,
    )

    response = JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            "error": "INTERNAL_SERVER_ERROR",
            "message": "Error interno del servidor",
            "status_code": status.HTTP_500_INTERNAL_SERVER_ERROR,
        },
    )
    for k, v in _cors_headers(request).items():
        response.headers[k] = v
    return response


def register_exception_handlers(app) -> None:
    """
    Registra todos los manejadores de excepciones en la aplicación FastAPI

    Args:
        app: Instancia de FastAPI
    """
    app.add_exception_handler(BaseAPIException, base_exception_handler)
    app.add_exception_handler(RequestValidationError, validation_exception_handler)
    app.add_exception_handler(IntegrityError, integrity_error_handler)
    app.add_exception_handler(SQLAlchemyError, sqlalchemy_error_handler)
    app.add_exception_handler(Exception, general_exception_handler)