""" Excepciones personalizadas para la aplicación. Define excepciones específicas del dominio y handlers para convertirlas en respuestas HTTP apropiadas. """ from typing import Any, Dict, Optional from fastapi import HTTPException, Request, status from fastapi.responses import JSONResponse class AdanException(Exception): """Excepción base para todas las excepciones de la aplicación.""" def __init__( self, message: str, code: str = "ADAN_ERROR", details: Optional[Dict[str, Any]] = None, ): self.message = message self.code = code self.details = details or {} super().__init__(self.message) class NotFoundError(AdanException): """Recurso no encontrado.""" def __init__( self, resource: str, identifier: Any = None, details: Optional[Dict[str, Any]] = None, ): message = f"{resource} no encontrado" if identifier: message = f"{resource} con id '{identifier}' no encontrado" super().__init__(message, "NOT_FOUND", details) self.resource = resource self.identifier = identifier class AlreadyExistsError(AdanException): """El recurso ya existe.""" def __init__( self, resource: str, field: str, value: Any, details: Optional[Dict[str, Any]] = None, ): message = f"{resource} con {field}='{value}' ya existe" super().__init__(message, "ALREADY_EXISTS", details) self.resource = resource self.field = field self.value = value class ValidationError(AdanException): """Error de validación de datos.""" def __init__( self, message: str, field: Optional[str] = None, details: Optional[Dict[str, Any]] = None, ): super().__init__(message, "VALIDATION_ERROR", details) self.field = field class AuthenticationError(AdanException): """Error de autenticación.""" def __init__( self, message: str = "Credenciales inválidas", details: Optional[Dict[str, Any]] = None, ): super().__init__(message, "AUTHENTICATION_ERROR", details) class AuthorizationError(AdanException): """Error de autorización (permisos insuficientes).""" def __init__( self, message: str = "No tiene permisos para realizar esta acción", details: Optional[Dict[str, Any]] = None, ): super().__init__(message, "AUTHORIZATION_ERROR", details) class ExternalServiceError(AdanException): """Error al comunicarse con un servicio externo.""" def __init__( self, service: str, message: str, details: Optional[Dict[str, Any]] = None, ): full_message = f"Error en servicio {service}: {message}" super().__init__(full_message, "EXTERNAL_SERVICE_ERROR", details) self.service = service class GeocercaViolationError(AdanException): """Violación de geocerca detectada.""" def __init__( self, geocerca_id: int, geocerca_nombre: str, tipo_violacion: str, # 'entrada' o 'salida' vehiculo_id: int, details: Optional[Dict[str, Any]] = None, ): message = f"Vehículo {vehiculo_id} {tipo_violacion} de geocerca '{geocerca_nombre}'" super().__init__(message, "GEOCERCA_VIOLATION", details) self.geocerca_id = geocerca_id self.geocerca_nombre = geocerca_nombre self.tipo_violacion = tipo_violacion self.vehiculo_id = vehiculo_id class SpeedLimitExceededError(AdanException): """Límite de velocidad excedido.""" def __init__( self, vehiculo_id: int, velocidad: float, limite: float, details: Optional[Dict[str, Any]] = None, ): message = f"Vehículo {vehiculo_id} excedió límite de velocidad: {velocidad} km/h (límite: {limite} km/h)" super().__init__(message, "SPEED_LIMIT_EXCEEDED", details) self.vehiculo_id = vehiculo_id self.velocidad = velocidad self.limite = limite class DeviceConnectionError(AdanException): """Error de conexión con dispositivo.""" def __init__( self, dispositivo_id: int, message: str, details: Optional[Dict[str, Any]] = None, ): full_message = f"Error de conexión con dispositivo {dispositivo_id}: {message}" super().__init__(full_message, "DEVICE_CONNECTION_ERROR", details) self.dispositivo_id = dispositivo_id class VideoStreamError(AdanException): """Error con stream de video.""" def __init__( self, camara_id: int, message: str, details: Optional[Dict[str, Any]] = None, ): full_message = f"Error de video en cámara {camara_id}: {message}" super().__init__(full_message, "VIDEO_STREAM_ERROR", details) self.camara_id = camara_id class MaintenanceRequiredError(AdanException): """Mantenimiento requerido para el vehículo.""" def __init__( self, vehiculo_id: int, tipo_mantenimiento: str, details: Optional[Dict[str, Any]] = None, ): message = f"Vehículo {vehiculo_id} requiere mantenimiento: {tipo_mantenimiento}" super().__init__(message, "MAINTENANCE_REQUIRED", details) self.vehiculo_id = vehiculo_id self.tipo_mantenimiento = tipo_mantenimiento class DatabaseError(AdanException): """Error de base de datos.""" def __init__( self, operation: str, message: str, details: Optional[Dict[str, Any]] = None, ): full_message = f"Error de base de datos en {operation}: {message}" super().__init__(full_message, "DATABASE_ERROR", details) self.operation = operation # ============================================================================ # Exception Handlers para FastAPI # ============================================================================ async def adan_exception_handler(request: Request, exc: AdanException) -> JSONResponse: """Handler para excepciones base de Adan.""" status_code = status.HTTP_500_INTERNAL_SERVER_ERROR if isinstance(exc, NotFoundError): status_code = status.HTTP_404_NOT_FOUND elif isinstance(exc, AlreadyExistsError): status_code = status.HTTP_409_CONFLICT elif isinstance(exc, ValidationError): status_code = status.HTTP_422_UNPROCESSABLE_ENTITY elif isinstance(exc, AuthenticationError): status_code = status.HTTP_401_UNAUTHORIZED elif isinstance(exc, AuthorizationError): status_code = status.HTTP_403_FORBIDDEN elif isinstance(exc, ExternalServiceError): status_code = status.HTTP_502_BAD_GATEWAY return JSONResponse( status_code=status_code, content={ "error": { "code": exc.code, "message": exc.message, "details": exc.details, } }, ) async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse: """Handler para HTTPException estándar de FastAPI.""" return JSONResponse( status_code=exc.status_code, content={ "error": { "code": "HTTP_ERROR", "message": exc.detail, "details": {}, } }, ) async def general_exception_handler(request: Request, exc: Exception) -> JSONResponse: """Handler para excepciones no capturadas.""" # En producción, loguear el error completo pero no exponerlo al cliente import traceback traceback.print_exc() return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={ "error": { "code": "INTERNAL_ERROR", "message": "Error interno del servidor", "details": {}, } }, ) def register_exception_handlers(app) -> None: """ Registra los handlers de excepciones en la aplicación FastAPI. Args: app: Instancia de FastAPI. """ app.add_exception_handler(AdanException, adan_exception_handler) app.add_exception_handler(HTTPException, http_exception_handler) app.add_exception_handler(Exception, general_exception_handler)