ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: refactor invoice compliance fields to use foreign keys and enhance validation logic

Browse Source
This commit is contained in:
AlexeerCT
2026-01-06 18:01:26 -06:00
parent 12ec8c6dfb
commit 9fe07e78a6
14 changed files with 704 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
backend/api/v1/modules/a76/invoices/common/common_validators.py Normal file
View File

@@ -0,0 +1,28 @@
from core.exceptions import ErrorCollector
from .. import models
from sqlalchemy.orm import Session
def invoice_exists(
db: Session,
invoice_number: str,
tenant_id: int,
company_id: int,
errors: ErrorCollector
) -> bool:
invoice_exists = (
db.query(models.InvoiceHeader.id)
.filter(
models.InvoiceHeader.invoice_number == invoice_number,
models.InvoiceHeader.tenant_id == tenant_id,
models.InvoiceHeader.company_id == company_id,
)
.first()
)
if not invoice_exists:
errors.add_duplicate_error(
"invoice_number",
invoice_number,
f"Ya existe una factura con el número '{invoice_number}'",
)

1
backend/api/v1/modules/a76/invoices/common/create_validators.py Normal file
View File

@@ -0,0 +1 @@

15
backend/api/v1/modules/a76/invoices/common/mappers.py Normal file
View File

@@ -0,0 +1,15 @@
""" """
def clean_dict(data_dict: dict) -> dict:
cleaned = {}
for key, value in data_dict.items():
if isinstance(value, str) and not value.strip():
cleaned[key] = None
elif value == 0 and (key.endswith('_id') or key == 'remesa'):
cleaned[key] = None
else:
cleaned[key] = value
return cleaned

9
backend/api/v1/modules/a76/invoices/imports/temporary/validators/common.py Normal file
View File

@@ -0,0 +1,9 @@
from sqlalchemy.orm import Session
from .... import schemas
from core.exceptions import ErrorCollector
def validate_common(db: Session, invoice: schemas.InvoiceTemporaryCreate, tenant_id: int, company_id: int, errors: ErrorCollector):
if invoice.compliance_mx.pedimento_id:
len()

82
backend/api/v1/modules/a76/invoices/imports/temporary/validators/create.py Normal file
View File

@@ -0,0 +1,82 @@
from sqlalchemy.orm import Session
from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
from core.exceptions import ErrorCollector
from ....schemas import InvoiceHeaderCreate
from .common import validate_common
def validate_create(db: Session, invoice: InvoiceHeaderCreate, tenant_id: int, company_id: int, errors: ErrorCollector) -> None:
""" Valida la creación de una nueva factura de importe temporal """
if not invoice.invoice_number:
errors.add_required_error("invoice_number")
if not invoice.invoice_date:
errors.add_required_error("invoice_date")
if not invoice.document_type:
errors.add_required_error("document_type")
if not invoice.compliance_mx.provider_id:
errors.add_required_error("compliance_mx.provider_id")
if not invoice.compliance_mx.sold_to_id:
errors.add_required_error("compliance_mx.sold_to_id")
if not invoice.compliance_mx.shipped_to_id:
errors.add_required_error("compliance_mx.shipped_to_id")
if not invoice.compliance_mx.customs_broker_id:
errors.add_required_error("compliance_mx.customs_broker_id")
if not invoice.compliance_mx.aduana:
errors.add_required_error("compliance_mx.aduana")
if errors.has_errors():
"""Se retorna por que hay campos obligatiorios para las validaciones que tienen que ser llenados"""
return
validate_common(db, invoice, tenant_id, company_id, errors)
if errors.has_errors():
"""Se retorna por que fallaron las validaciones generales"""
return
if not invoice.compliance_mx.pedimento_id:
invoice.compliance_mx.remesa = None
if not invoice.financials.exchange_rate:
invoice.financials.exchange_rate = db.query(ExchangeRate.value).filter(ExchangeRate.date == invoice.invoice_date).scalar()
invoice.document_type = (invoice.document_type or "").upper()
if not invoice.logistics.transport_type:
invoice.logistics.transport_type = "none"
if invoice.logistics.transport_type == "none" and invoice.logistics.transport_num:
invoice.logistics.transport_num = None
if not invoice.financials.currency:
invoice.financials.currency = "foreign"
if invoice.financials.currency == "local":
invoice.financials.currency_type = "MXN"
elif invoice.financials.currency_type == "foreign":
invoice.financials.currency = "USD"
elif invoice.financials.currency_type == "manual":
invoice.financials.currency_type = invoice.financials.currency_type.upper()
invoice.logistics.incoterm = (invoice.logistics.incoterm or "").upper()
if not invoice.logistics.weight_type:
invoice.logistics.weight_type = "kgs"

2
backend/api/v1/modules/a76/invoices/imports/temporary/validators/update.py Normal file
View File

@@ -0,0 +1,2 @@
def validate_update():
pass

6
backend/api/v1/modules/a76/invoices/models.py
View File

@@ -136,9 +136,9 @@ class InvoiceComplianceMx(Base, TenantScopedMixin, TimestampMixin):
invoice_id: Mapped[int] = mapped_column(ForeignKey("a76.invoice_header.id"), primary_key=True)
# Core Customs Data
pedimento: Mapped[Optional[str]] = mapped_column(String(19)) # PEDIMENTO/PEDIMENTOIMPO/EXPO
pedimento_code: Mapped[Optional[str]] = mapped_column(String(5)) # PEDIMENTOR1
pedimento_k1: Mapped[Optional[str]] = mapped_column(String(15)) # PEDIMENTOK1
pedimento_id: Mapped[Optional[str]] = mapped_column(ForeignKey("a76.pedimentos.id")) # PEDIMENTO/PEDIMENTOIMPO/EXPO
pedimento_r1: Mapped[Optional[str]] = mapped_column(ForeignKey("a76.pedimentos.id")) # PEDIMENTOR1
pedimento_k1: Mapped[Optional[str]] = mapped_column(ForeignKey("a76.pedimentos.id")) # PEDIMENTOK1
remesa: Mapped[Optional[int]] = mapped_column(Integer) # REMESA
aduana: Mapped[Optional[str]] = mapped_column(ForeignKey("public.customs_sections.customs_code")) # ADUANA_CRUCE
port_of_entry: Mapped[Optional[str]] = mapped_column(String(6)) # PUERTOENTRADA / Puerto de entrada

15
backend/api/v1/modules/a76/invoices/schemas.py
View File

@@ -84,15 +84,14 @@ class InvoiceHeaderBase(BaseModel):
class InvoiceComplianceMxBase(BaseModel):
"""Base fields for Compliance MX"""
pedimento: Optional[str] = Field(
None, max_length=19, description="Pedimento number")
pedimento_code: Optional[str] = Field(
None, max_length=5, description="Pedimento code (R1)")
pedimento_k1: Optional[str] = Field(
None, max_length=15, description="Pedimento K1")
pedimento_id: Optional[int] = Field(
None, description="Pedimento id")
pedimento_r1: Optional[int] = Field(
None, description="Pedimento id (R1)")
pedimento_k1: Optional[int] = Field(
None, description="Pedimento id (K1)")
remesa: Optional[int] = Field(None, description="Remesa")
aduana: Optional[str] = Field(
None, max_length=5, description="Customs office")
aduana: Optional[str] = Field(None, max_length=5, description="Customs office")
port_of_entry: Optional[str] = Field(
None, max_length=6, description="Port of entry")
destination: Optional[str] = Field(

140
backend/api/v1/modules/a76/invoices/services.py
View File

@@ -1,15 +1,22 @@
import traceback
from typing import Optional, List, Tuple
from sqlalchemy.orm import Session
from sqlalchemy import and_
from core.exceptions import ErrorCollector, DuplicateResourceException
from .common.mappers import clean_dict
from .imports.temporary.validators.create import validate_create
from .imports.temporary.validators.update import validate_update
from .common.common_validators import invoice_exists
from . import models, schemas
class InvoiceService:
"""Service for Invoice Header operations"""
@staticmethod
def get_by_id(db: Session, invoice_id: int, tenant_id: int, company_id: int) -> Optional[models.InvoiceHeader]:
def get_by_id(
db: Session, invoice_id: int, tenant_id: int, company_id: int
) -> Optional[models.InvoiceHeader]:
"""Get an invoice by ID with tenant/company validation"""
return (
db.query(models.InvoiceHeader)
@@ -39,25 +46,32 @@ class InvoiceService:
# Apply filters if provided
if filters:
if filters.get("status"):
query = query.filter(
models.InvoiceHeader.status == filters["status"])
query = query.filter(models.InvoiceHeader.status == filters["status"])
if filters.get("operation_type"):
query = query.filter(
models.InvoiceHeader.operation_type == filters["operation_type"])
models.InvoiceHeader.operation_type == filters["operation_type"]
)
if filters.get("invoice_type"):
query = query.filter(
models.InvoiceHeader.invoice_type == filters["invoice_type"])
models.InvoiceHeader.invoice_type == filters["invoice_type"]
)
if filters.get("invoice_number"):
query = query.filter(models.InvoiceHeader.invoice_number.ilike(
f"%{filters['invoice_number']}%"))
query = query.filter(
models.InvoiceHeader.invoice_number.ilike(
f"%{filters['invoice_number']}%"
)
)
if filters.get("pedimento"):
query = query.join(models.InvoiceComplianceMx).filter(
models.InvoiceComplianceMx.pedimento.ilike(
f"%{filters['pedimento']}%")
f"%{filters['pedimento']}%"
)
)
if not filters.get("invoice_type") and filters.get("operation_type") == "exp":
query = query.filter(
models.InvoiceHeader.operation_type != "REPAR")
if (
not filters.get("invoice_type")
and filters.get("operation_type") == "exp"
):
query = query.filter(models.InvoiceHeader.operation_type != "REPAR")
total = query.count()
items = query.offset(skip).limit(limit).all()
@@ -68,30 +82,19 @@ class InvoiceService:
db: Session,
invoice_data: schemas.InvoiceHeaderCreate,
tenant_id: int,
company_id: int
company_id: int,
) -> models.InvoiceHeader:
"""Create a new invoice with all related data"""
def clean_dict(data_dict: dict) -> dict:
cleaned = {}
for key, value in data_dict.items():
if key == 'customs_agent':
key = 'customs_broker_id'
elif key == 'provider':
key = 'provider_id'
if isinstance(value, str) and not value.strip():
cleaned[key] = None
elif value == 0 and (key.endswith('_id') or key == 'remesa'):
cleaned[key] = None
else:
cleaned[key] = value
return cleaned
# Validaciones con ErrorCollector
errors = ErrorCollector()
# Validar si la factura ya existe
invoice_exists(db, invoice_data.invoice_number, tenant_id, company_id, errors)
validate_create(db, invoice_data, tenant_id, company_id, errors)
# Si hay errores, lanzar excepción
errors.raise_if_errors("Error al crear la factura")
try:
# Extract nested data
@@ -103,27 +106,32 @@ class InvoiceService:
# Create main invoice header
raw_invoice_dict = invoice_data.model_dump(
exclude={"compliance_mx", "financials",
"logistics", "details", "collections"}
exclude={
"compliance_mx",
"financials",
"logistics",
"details",
"collections",
}
)
invoice_dict = clean_dict(raw_invoice_dict)
invoice_dict["tenant_id"] = tenant_id
invoice_dict["company_id"] = company_id
new_invoice = models.InvoiceHeader(**invoice_dict)
db.add(new_invoice)
db.flush() # Flush to get the invoice ID
# Create compliance_mx if provided
if compliance_data:
raw_comp_dict = compliance_data.model_dump()
# Pasamos los datos por la lavadora para arreglar pedimento, aduana, etc.
compliance_dict = clean_dict(raw_comp_dict)
compliance_dict["invoice_id"] = new_invoice.id
compliance_dict["tenant_id"] = tenant_id
compliance_dict["company_id"] = company_id
new_compliance = models.InvoiceComplianceMx(**compliance_dict)
db.add(new_compliance)
@@ -131,11 +139,11 @@ class InvoiceService:
if financials_data:
raw_fin_dict = financials_data.model_dump()
financials_dict = clean_dict(raw_fin_dict)
financials_dict["invoice_id"] = new_invoice.id
financials_dict["tenant_id"] = tenant_id
financials_dict["company_id"] = company_id
new_financials = models.InvoiceFinancials(**financials_dict)
db.add(new_financials)
@@ -143,7 +151,7 @@ class InvoiceService:
for logistics_item in logistics_data:
raw_log_dict = logistics_item.model_dump()
logistics_dict = clean_dict(raw_log_dict)
logistics_dict["invoice_id"] = new_invoice.id
logistics_dict["tenant_id"] = tenant_id
logistics_dict["company_id"] = company_id
@@ -154,7 +162,7 @@ class InvoiceService:
for detail_item in details_data:
raw_det_dict = detail_item.model_dump()
detail_dict = clean_dict(raw_det_dict)
detail_dict["invoice_id"] = new_invoice.id
detail_dict["tenant_id"] = tenant_id
detail_dict["company_id"] = company_id
@@ -165,7 +173,7 @@ class InvoiceService:
for collection_item in collections_data:
raw_col_dict = collection_item.model_dump()
collection_dict = clean_dict(raw_col_dict)
collection_dict["invoice_id"] = new_invoice.id
collection_dict["tenant_id"] = tenant_id
collection_dict["company_id"] = company_id
@@ -180,7 +188,7 @@ class InvoiceService:
db.rollback()
print("\n\n🔥 ERROR AL GUARDAR FACTURA 🔥")
print(f"Error: {str(e)}")
traceback.print_exc() # Esto imprime el error real en la consola
traceback.print_exc() # Esto imprime el error real en la consola
print("--------------------------------\n")
raise e
@@ -190,20 +198,24 @@ class InvoiceService:
invoice_id: int,
tenant_id: int,
invoice_data: schemas.InvoiceHeaderUpdate,
company_id: int
company_id: int,
) -> Optional[models.InvoiceHeader]:
# ... (El resto de tu código update se queda igual) ...
# (Te recomiendo implementar clean_dict aquí también si tienes problemas al editar)
invoice = InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
invoice = InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
if not invoice:
return None
# Update main invoice header fields
update_dict = invoice_data.model_dump(
exclude={"compliance_mx", "financials",
"logistics", "details", "collections"},
exclude_unset=True
exclude={
"compliance_mx",
"financials",
"logistics",
"details",
"collections",
},
exclude_unset=True,
)
for key, value in update_dict.items():
setattr(invoice, key, value)
@@ -211,15 +223,21 @@ class InvoiceService:
# Update compliance_mx if provided
if invoice_data.compliance_mx is not None:
if invoice.compliance_mx:
for key, value in invoice_data.compliance_mx.model_dump(exclude_unset=True).items():
for key, value in invoice_data.compliance_mx.model_dump(
exclude_unset=True
).items():
# Parche rápido para update
if value == "": value = None
if value == "":
value = None
setattr(invoice.compliance_mx, key, value)
else:
compliance_dict = invoice_data.compliance_mx.model_dump()
# Aplicar limpieza manual si es necesario
if 'customs_agent' in compliance_dict: compliance_dict['customs_broker_id'] = compliance_dict.pop('customs_agent')
if "customs_agent" in compliance_dict:
compliance_dict["customs_broker_id"] = compliance_dict.pop(
"customs_agent"
)
compliance_dict["invoice_id"] = invoice.id
compliance_dict["tenant_id"] = tenant_id
compliance_dict["company_id"] = company_id
@@ -229,8 +247,11 @@ class InvoiceService:
# Update financials if provided
if invoice_data.financials is not None:
if invoice.financials:
for key, value in invoice_data.financials.model_dump(exclude_unset=True).items():
if value == "": value = None
for key, value in invoice_data.financials.model_dump(
exclude_unset=True
).items():
if value == "":
value = None
setattr(invoice.financials, key, value)
else:
financials_dict = invoice_data.financials.model_dump()
@@ -247,10 +268,9 @@ class InvoiceService:
@staticmethod
def delete(db: Session, invoice_id: int, tenant_id: int, company_id: int) -> bool:
"""Delete an invoice and all related data (cascade delete)"""
invoice = InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
invoice = InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
if invoice:
db.delete(invoice)
db.commit()
return True
return False
return False

2
backend/api/v1/modules/a76/items/models.py
View File

@@ -106,7 +106,7 @@ class SubassemblyEntry(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
remission_line: Mapped[int] = mapped_column(Integer) # LINEAREMISION
exit_invoice: Mapped[Optional[str]] = mapped_column(
String(15)) # FACTURASALIDA
String(15)) # FACTURASALIDA
exit_line: Mapped[Optional[int]] = mapped_column(Integer) # LINEASALIDA
# ============================================================================

2
backend/api/v1/modules/public/reference_data/currency_types/seed.py
View File

@@ -55,7 +55,7 @@ seed = [
("LTT", "LITAS", "LITUANIA"),
("LYD", "DINAR", "LIBIA"),
("MAD", "DIRHAM", "MARRUECOS"),
("MXP", "PESO", "MEXICO"),
("MXN", "PESO", "MEXICO"),
("MYR", "RINGGIT", "MALASIA"),
("NGN", "NAIRA", "NIGERIA (FED)"),
("NIC", "CORDOBA", "NICARAGUA"),

175
backend/core/error_handlers.py Normal file
View File

@@ -0,0 +1,175 @@
"""
Manejadores globales de excepciones para FastAPI
"""
import logging
from typing import Any, Dict
from fastapi import Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from .exceptions import BaseAPIException
logger = logging.getLogger(__name__)
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,
},
)
return JSONResponse(
status_code=exc.status_code,
content=exc.to_dict(),
)
async def validation_exception_handler(
request: Request,
exc: RequestValidationError,
) -> JSONResponse:
"""
Manejador para errores de validación de Pydantic/FastAPI
"""
errors = []
for error in exc.errors():
field = ".".join(str(loc) for loc in error["loc"] if loc != "body")
errors.append(
{
"field": field,
"message": error["msg"],
"type": error["type"],
}
)
logger.warning(
f"Validation Error en {request.url.path}",
extra={"errors": errors},
)
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={
"error": "VALIDATION_ERROR",
"message": "Error de validación en los datos recibidos",
"status_code": status.HTTP_422_UNPROCESSABLE_ENTITY,
"errors": errors,
},
)
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,
},
)
# Intentar extraer información útil del error
error_message = "Error de integridad en la base de datos"
orig_msg = str(exc.orig).lower()
if "unique constraint" in orig_msg or "duplicate key" in orig_msg:
error_message = "El registro ya existe. Verifica los campos únicos."
elif "foreign key" in orig_msg:
error_message = "Referencia inválida a otro registro."
elif "not null" in orig_msg:
error_message = "Falta un campo requerido."
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content={
"error": "DATABASE_INTEGRITY_ERROR",
"message": error_message,
"status_code": status.HTTP_409_CONFLICT,
},
)
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,
)
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={
"error": "DATABASE_ERROR",
"message": "Error en la operación de base de datos",
"status_code": status.HTTP_500_INTERNAL_SERVER_ERROR,
},
)
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,
)
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={
"error": "INTERNAL_SERVER_ERROR",
"message": "Error interno del servidor",
"status_code": status.HTTP_500_INTERNAL_SERVER_ERROR,
},
)
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(RequestValidationError, 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)
logger.info("Exception handlers registered successfully")

290
backend/core/exceptions.py Normal file
View File

@@ -0,0 +1,290 @@
"""
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,
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 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, 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", "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, "DUPLICATE", 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}", "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, "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()})"

12
backend/main.py
View File

@@ -8,6 +8,7 @@ import logging
from api.v1.router import router as api_v1_router
from core.config import settings
from core.database import init_db
from core.error_handlers import register_exception_handlers
from core.middleware import (
LicenseValidationMiddleware,
RequestLoggingMiddleware,
@@ -16,8 +17,12 @@ from core.middleware import (
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from api.v1.modules.a76.items.models import Item # Importar rutas para registrar con el router
from api.v1.modules.a76.items.series.models import Serie # Importar modelos para registrar con SQLAlchemy
from api.v1.modules.a76.items.models import (
Item,
) # Importar rutas para registrar con el router
from api.v1.modules.a76.items.series.models import (
Serie,
) # Importar modelos para registrar con SQLAlchemy
# Configurar logging
logging.basicConfig(
@@ -37,6 +42,9 @@ app = FastAPI(
openapi_url="/api/openapi.json" if settings.DEBUG else None,
)
# Registrar manejadores de excepciones
register_exception_handlers(app)
# Inicializar la base de datos
@app.on_event("startup")