358 lines
12 KiB
Python
358 lines
12 KiB
Python
"""
|
|
Manejadores globales de excepciones para FastAPI
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any, Dict
|
|
|
|
from fastapi import Request, status, HTTPException
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi.exceptions import RequestValidationError
|
|
from pydantic import ValidationError
|
|
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
|
|
|
|
|
|
# Mapa de campos técnicos a nombres legibles en español
|
|
_FIELD_LABELS: Dict[str, str] = {
|
|
"broker_key": "Clave del Agente",
|
|
"license": "Patente",
|
|
"tax_id": "RFC",
|
|
"personal_id": "CURP",
|
|
"email": "Correo Electrónico",
|
|
"phone": "Teléfono",
|
|
"fax": "Fax",
|
|
"contact": "Nombre de Contacto",
|
|
"name": "Nombre / Razón Social",
|
|
"address": "Dirección",
|
|
"postal_code": "Código Postal",
|
|
"city": "Ciudad",
|
|
"state": "Estado",
|
|
"country": "País",
|
|
# Partes (A76)
|
|
"unit_cost": "Costo Unitario",
|
|
"unit_weight": "Peso Unitario",
|
|
"sector": "Sector",
|
|
"fraction_type": "Tipo de tarifa",
|
|
}
|
|
|
|
_FIELD_PATTERN_MESSAGES: Dict[str, str] = {
|
|
"broker_key": "La Clave del Agente solo puede contener letras y números (máx. 5 caracteres).",
|
|
"license": "La Patente debe ser un número entre 1 y 9999 (no puede ser 0 ni contener letras).",
|
|
"tax_id": "El RFC no tiene el formato correcto. Ejemplo válido: XAXX010101000.",
|
|
"personal_id": "La CURP no tiene el formato correcto. Debe tener 18 caracteres alfanuméricos.",
|
|
"email": "El correo electrónico no tiene un formato válido. Ejemplo: usuario@dominio.com.",
|
|
"phone": "El teléfono solo puede contener dígitos, espacios y los símbolos: +, -, (, ).",
|
|
"contact": "El nombre de contacto contiene caracteres no permitidos. Use solo letras, números y puntuación básica.",
|
|
# Partes (A76)
|
|
"sector": "El Sector solo puede contener números, máximo 8 dígitos (sin espacios ni caracteres especiales).",
|
|
}
|
|
|
|
|
|
def _friendly_message(field_key: str, error_type: str) -> str:
|
|
"""Devuelve un mensaje de error legible en español según el campo y tipo de error."""
|
|
if error_type in ("greater_than_equal",):
|
|
if field_key in ("unit_cost", "unit_weight"):
|
|
return f"El campo '{_FIELD_LABELS.get(field_key, field_key)}' no puede ser negativo."
|
|
return f"El campo '{_FIELD_LABELS.get(field_key, field_key)}' debe ser mayor o igual a 0."
|
|
if error_type in ("string_pattern_mismatch", "value_error"):
|
|
return _FIELD_PATTERN_MESSAGES.get(
|
|
field_key,
|
|
f"El campo '{_FIELD_LABELS.get(field_key, field_key)}' contiene un valor con formato inválido.",
|
|
)
|
|
if error_type in ("decimal_parsing", "decimal_type", "float_parsing", "float_type", "int_parsing", "int_type"):
|
|
return (
|
|
f"El campo '{_FIELD_LABELS.get(field_key, field_key)}' debe ser numérico. "
|
|
"Si no aplica, déjelo vacío."
|
|
)
|
|
if error_type in ("literal_error",):
|
|
if field_key == "fraction_type":
|
|
return "El campo 'Tipo de tarifa' es inválido. Seleccione una opción predefinida."
|
|
return f"El campo '{_FIELD_LABELS.get(field_key, field_key)}' contiene una opción inválida."
|
|
if error_type == "string_too_long":
|
|
return f"El campo '{_FIELD_LABELS.get(field_key, field_key)}' excede la longitud máxima permitida."
|
|
if error_type == "string_too_short":
|
|
return f"El campo '{_FIELD_LABELS.get(field_key, field_key)}' es demasiado corto."
|
|
if error_type in ("missing", "value_error.missing"):
|
|
return f"El campo '{_FIELD_LABELS.get(field_key, field_key)}' es obligatorio."
|
|
return f"El campo '{_FIELD_LABELS.get(field_key, field_key)}' contiene un valor inválido."
|
|
|
|
|
|
async def validation_exception_handler(
|
|
request: Request,
|
|
exc: RequestValidationError,
|
|
) -> JSONResponse:
|
|
"""
|
|
Manejador para errores de validación de Pydantic/FastAPI.
|
|
Devuelve mensajes legibles en español.
|
|
"""
|
|
errors = []
|
|
for error in exc.errors():
|
|
loc_parts = [str(loc) for loc in error["loc"] if loc != "body"]
|
|
field = ".".join(loc_parts)
|
|
field_key = loc_parts[-1] if loc_parts else ""
|
|
|
|
errors.append(
|
|
{
|
|
"field": field,
|
|
"message": _friendly_message(field_key, error["type"]),
|
|
"type": error["type"],
|
|
}
|
|
)
|
|
|
|
print(f"DEBUG REQUEST VALIDATION ERRORS: {errors}")
|
|
logger.warning(
|
|
f"Validation Error en {request.url.path}",
|
|
extra={"errors": errors},
|
|
)
|
|
|
|
summary = (
|
|
errors[0]["message"]
|
|
if len(errors) == 1
|
|
else f"Hay {len(errors)} errores de validación: " + " | ".join(e["message"] for e in errors)
|
|
)
|
|
|
|
response = JSONResponse(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
content={
|
|
"error": "VALIDATION_ERROR",
|
|
"message": summary,
|
|
"status_code": status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
"errors": errors,
|
|
},
|
|
)
|
|
for k, v in _cors_headers(request).items():
|
|
response.headers[k] = v
|
|
return response
|
|
|
|
|
|
import traceback
|
|
|
|
async def inner_validation_exception_handler(
|
|
request: Request,
|
|
exc: ValidationError,
|
|
) -> JSONResponse:
|
|
"""
|
|
Manejador para errores de validación de Pydantic lanzados internamente (como en tenant_crud_routes).
|
|
"""
|
|
traceback.print_exc()
|
|
errors = []
|
|
for error in exc.errors():
|
|
loc_parts = [str(loc) for loc in error["loc"] if loc != "body"]
|
|
field = ".".join(loc_parts)
|
|
field_key = loc_parts[-1] if loc_parts else ""
|
|
|
|
errors.append(
|
|
{
|
|
"field": field,
|
|
"message": _friendly_message(field_key, error["type"]),
|
|
"type": error["type"],
|
|
}
|
|
)
|
|
|
|
print(f"DEBUG VALIDATION ERRORS: {errors}")
|
|
logger.warning(
|
|
f"Inner Validation Error en {request.url.path}",
|
|
extra={"errors": errors},
|
|
)
|
|
|
|
summary = (
|
|
errors[0]["message"]
|
|
if len(errors) == 1
|
|
else f"Hay {len(errors)} errores de validación: " + " | ".join(e["message"] for e in errors)
|
|
)
|
|
|
|
response = JSONResponse(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
content={
|
|
"error": "VALIDATION_ERROR",
|
|
"message": summary,
|
|
"status_code": status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
"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,
|
|
},
|
|
)
|
|
|
|
orig_msg = str(exc.orig).lower()
|
|
|
|
# Check for unique/duplicate key violations (English and Spanish)
|
|
if any(kw in orig_msg for kw in ["unique constraint", "duplicate key", "duplicada", "unicidad", "ya existe"]):
|
|
error_message = "El registro ya existe. Verifica los campos únicos (Año, Aduana, Patente, Número, etc.)."
|
|
# Check for foreign key violations (English and Spanish)
|
|
elif any(kw in orig_msg for kw in ["foreign key", "foránea", "referencia"]):
|
|
error_message = "Referencia inválida a otro registro. Verifica las categorías y catálogos seleccionados."
|
|
# Check for not null violations (English and Spanish)
|
|
elif any(kw in orig_msg for kw in ["not null", "no nulo", "valor nulo"]):
|
|
error_message = "Falta un campo requerido. Asegúrate de llenar todos los datos obligatorios."
|
|
else:
|
|
error_message = "Error de integridad en la base de datos"
|
|
|
|
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": f"Error en la operación de base de datos: {str(exc)}",
|
|
"status_code": status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
},
|
|
)
|
|
for k, v in _cors_headers(request).items():
|
|
response.headers[k] = v
|
|
return response
|
|
|
|
|
|
async def http_exception_handler(
|
|
request: Request,
|
|
exc: HTTPException,
|
|
) -> JSONResponse:
|
|
"""
|
|
Manejador para HTTPException de FastAPI
|
|
"""
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content={
|
|
"error": "HTTP_ERROR",
|
|
"message": exc.detail,
|
|
"status_code": exc.status_code,
|
|
},
|
|
)
|
|
|
|
|
|
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": f"Error interno del servidor: {str(exc)}",
|
|
"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(HTTPException, http_exception_handler)
|
|
app.add_exception_handler(RequestValidationError, validation_exception_handler)
|
|
app.add_exception_handler(ValidationError, inner_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)
|
|
|