""" Sistema centralizado de excepciones personalizadas """ from typing import Optional, List, Dict, Any from fastapi import status class BaseAPIException(Exception): """Excepción base para todas las excepciones de la API""" def __init__( self, message: str, status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR, errors: Optional[List[Dict[str, Any]]] = None, error_code: Optional[str] = None, ): self.message = message self.status_code = status_code self.errors = errors or [] self.error_code = error_code or self.__class__.__name__ super().__init__(self.message) def to_dict(self) -> Dict[str, Any]: """Convierte la excepción a un diccionario para respuesta JSON""" response = { "error": self.error_code, "message": self.message, "status_code": self.status_code, } if self.errors: response["errors"] = self.errors return response class ValidationException(BaseAPIException): """Excepción para errores de validación""" def __init__( self, message: str = "Error de validación", errors: Optional[List[Dict[str, Any]]] = None, ): super().__init__( message=message, status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, errors=errors, error_code="VALIDATION_ERROR", ) class DuplicateResourceException(BaseAPIException): """Excepción cuando se intenta crear un recurso duplicado""" def __init__( self, resource: str, identifier: str, message: Optional[str] = None, ): self.resource = resource self.identifier = identifier final_message = ( message or f"{resource} con identificador '{identifier}' ya existe" ) super().__init__( message=final_message, status_code=status.HTTP_409_CONFLICT, error_code="DUPLICATE_RESOURCE", ) class ResourceNotFoundException(BaseAPIException): """Excepción cuando no se encuentra un recurso""" def __init__( self, resource: str, identifier: str, message: Optional[str] = None, ): self.resource = resource self.identifier = identifier final_message = ( message or f"{resource} con identificador '{identifier}' no encontrado" ) super().__init__( message=final_message, status_code=status.HTTP_404_NOT_FOUND, error_code="RESOURCE_NOT_FOUND", ) class UnauthorizedException(BaseAPIException): """Excepción para errores de autenticación""" def __init__(self, message: str = "No autorizado"): super().__init__( message=message, status_code=status.HTTP_401_UNAUTHORIZED, error_code="UNAUTHORIZED", ) class ForbiddenException(BaseAPIException): """Excepción para errores de permisos""" def __init__(self, message: str = "Acceso prohibido"): super().__init__( message=message, status_code=status.HTTP_403_FORBIDDEN, error_code="FORBIDDEN", ) class BusinessRuleException(BaseAPIException): """Excepción para errores de reglas de negocio""" def __init__( self, message: str, errors: Optional[List[Dict[str, Any]]] = None, ): super().__init__( message=message, status_code=status.HTTP_400_BAD_REQUEST, errors=errors, error_code="BUSINESS_RULE_ERROR", ) class DatabaseException(BaseAPIException): """Excepción para errores de base de datos""" def __init__(self, message: str = "Error en la base de datos"): super().__init__( message=message, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, error_code="DATABASE_ERROR", ) class ErrorCollector: """ Colector de errores para acumular múltiples errores de validación antes de lanzar una excepción Uso: collector = ErrorCollector() if not valid_email: collector.add_error("email", "Email inválido", "INVALID_EMAIL") if not valid_phone: collector.add_error("phone", "Teléfono inválido", "INVALID_PHONE") collector.raise_if_errors() # Lanza ValidationException si hay errores """ def __init__(self): self._errors: List[Dict[str, Any]] = [] def add_error( self, field: str, message: str, solution: Optional[List[str]], code: Optional[str] = None, value: Optional[Any] = None, ) -> "ErrorCollector": """ Agrega un error al colector Args: field: Campo donde ocurrió el error (ej: "invoice_number", "email") message: Mensaje descriptivo del error code: Código opcional del error (ej: "REQUIRED", "INVALID_FORMAT") value: Valor que causó el error (opcional) Returns: Self para permitir encadenamiento """ error = { "field": field, "message": message, } if solution: error["solution"] = solution if code: error["code"] = code if value is not None: error["value"] = value self._errors.append(error) return self def add_field_error( self, field: str, message: str, code: str = "INVALID", ) -> "ErrorCollector": """Atajo para agregar error de campo""" return self.add_error(field, message, solution=None, code=code) def add_required_error(self, field: str) -> "ErrorCollector": """Atajo para agregar error de campo requerido""" return self.add_error( field, f"El campo '{field}' es requerido", solution=None, code="REQUIRED" ) def add_duplicate_error( self, field: str, value: Any, message: Optional[str] = None, ) -> "ErrorCollector": """Atajo para agregar error de duplicado""" final_message = ( message or f"El valor '{value}' ya existe para el campo '{field}'" ) return self.add_error( field, final_message, solution=None, code="DUPLICATE", value=value ) def add_invalid_format_error( self, field: str, expected_format: str, ) -> "ErrorCollector": """Atajo para agregar error de formato inválido""" return self.add_error( field, f"Formato inválido. Se esperaba: {expected_format}", solution=None, code="INVALID_FORMAT", ) def add_range_error( self, field: str, min_value: Optional[Any] = None, max_value: Optional[Any] = None, ) -> "ErrorCollector": """Atajo para agregar error de rango""" if min_value is not None and max_value is not None: message = f"El valor debe estar entre {min_value} y {max_value}" elif min_value is not None: message = f"El valor debe ser mayor o igual a {min_value}" elif max_value is not None: message = f"El valor debe ser menor o igual a {max_value}" else: message = "Valor fuera de rango" return self.add_error(field, message, solution=None, code="OUT_OF_RANGE") def has_errors(self) -> bool: """Verifica si hay errores acumulados""" return len(self._errors) > 0 def get_errors(self) -> List[Dict[str, Any]]: """Obtiene la lista de errores""" return self._errors.copy() def get_error_count(self) -> int: """Obtiene el número de errores""" return len(self._errors) def clear(self) -> "ErrorCollector": """Limpia todos los errores""" self._errors.clear() return self def raise_if_errors( self, message: str = "Se encontraron errores de validación", ) -> None: """ Lanza ValidationException si hay errores acumulados Args: message: Mensaje principal de la excepción Raises: ValidationException: Si hay errores acumulados """ if self.has_errors(): raise ValidationException(message=message, errors=self._errors) def __bool__(self) -> bool: """Permite usar el colector en contextos booleanos""" return self.has_errors() def __len__(self) -> int: """Permite usar len() en el colector""" return self.get_error_count() def __repr__(self) -> str: return f"ErrorCollector(errors={self.get_error_count()})"