301 lines
8.7 KiB
Python
301 lines
8.7 KiB
Python
"""
|
|
Sistema centralizado de excepciones personalizadas para Anexo76
|
|
"""
|
|
|
|
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_ENTITY,
|
|
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()})"
|