feat: plantilla base workspace SaaS
This commit is contained in:
300
backend/core/exceptions.py
Normal file
300
backend/core/exceptions.py
Normal file
@@ -0,0 +1,300 @@
|
||||
"""
|
||||
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()})"
|
||||
Reference in New Issue
Block a user