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

feat(validation): implement comprehensive validation for item creation and updates

Browse Source
This commit is contained in:
Galindo97
2026-01-20 12:53:08 -06:00
parent ba353493d3
commit dd6c832674
6 changed files with 611 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
backend/api/v1/modules/a76/items/service.py
View File

@@ -16,6 +16,10 @@ from sqlalchemy import and_, or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, joinedload
from core.exceptions import ErrorCollector
from .imports.temporary.validators.create import validate_create
from .imports.temporary.validators.update import validate_update
from api.v1.modules.a76.items.line_items.schemas import LineItemCreate, LineItemUpdate
from .schemas import ItemCreate, ItemUpdate
@@ -27,6 +31,9 @@ from .line_descriptions.models import LineDescription
from .line_references.models import LineReference
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from .models import Item
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.general_catalogs.company.models import Company
logger = logging.getLogger(__name__)
@@ -151,6 +158,91 @@ class ItemService:
company_id: int,
) -> Item:
"""Create a new item with all related nested data (multiple lines)"""
# Validaciones con ErrorCollector
errors = ErrorCollector()
# Validar que la factura exista y no esté actualizada (si viene invoice_id)
if item_data.invoice_id:
invoice = db.query(InvoiceHeader).filter(
InvoiceHeader.id == item_data.invoice_id,
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id
).first()
if not invoice:
errors.add_error(
field="invoice_id",
message="La factura especificada no existe",
code="NOT_FOUND",
value=str(item_data.invoice_id)
)
else:
# Validar que la factura no esté actualizada (excepto RFCs especiales)
company = db.query(Company).filter(Company.id == company_id).first()
if company:
rfc_excepciones = ['TPI121217SF6', 'TCI170502858']
if company.rfc not in rfc_excepciones:
if invoice.is_updated:
errors.add_error(
field="invoice_id",
message="La factura está actualizada y no puede ser modificada",
code="INVOICE_UPDATED",
value=str(item_data.invoice_id)
)
# Validar cada line item que se va a crear
if item_data.lines:
for idx, line_data in enumerate(item_data.lines):
# Convertir a LineItemCreate para validar
line_create = LineItemCreate(**line_data.model_dump())
validate_create(db, line_create, tenant_id, company_id, errors)
# Validaciones adicionales específicas del negocio
# Validar apóstrofes en número de parte
if line_data.part_number_id and "'" in str(line_data.part_number_id):
errors.add_error(
field=f"lines[{idx}].part_number_id",
message=f"Advertencia: El Número de Parte contiene apóstrofes y serán omitidos",
code="WARNING_APOSTROPHE"
)
# Validar tipo de partida
if hasattr(line_data, 'item_type'):
tipo_partida = line_data.item_type
if tipo_partida and tipo_partida not in ['N', 'S']:
errors.add_error(
field=f"lines[{idx}].item_type",
message=f"Tipo de partida debe ser 'N' (Normal) o 'S' (Subpartida), recibido: '{tipo_partida}'",
code="INVALID_ITEM_TYPE",
value=str(tipo_partida)
)
# Si es subpartida (S), debe tener partida principal
if tipo_partida == 'S':
if not hasattr(line_data, 'main_line_id') or not line_data.main_line_id:
errors.add_error(
field=f"lines[{idx}].main_line_id",
message="Las subpartidas (tipo 'S') deben tener una partida principal",
code="MISSING_MAIN_LINE"
)
# Validar que el line_number sea consecutivo (si se especifica)
if hasattr(line_data, 'line_number') and line_data.line_number:
expected_line = idx + 1
if line_data.line_number != expected_line:
errors.add_error(
field=f"lines[{idx}].line_number",
message=f"Número de línea esperado: {expected_line}, recibido: {line_data.line_number}",
code="INVALID_LINE_SEQUENCE",
value=str(line_data.line_number)
)
# Si hay errores, lanzar excepción ANTES de intentar crear
errors.raise_if_errors("Error al crear el item")
try:
# Extract lines data
lines_data = item_data.lines or []
@@ -231,7 +323,7 @@ class ItemService:
# Create FA data if provided
if fa_data:
fa_dict = fa_data.model_dump()
fa_dict = fa_data.model_dump(exclude={"line_item_id"}) # Exclude line_item_id from DTO
fa_dict["id"] = db_line.id # FA table uses same ID as line item
fa_dict["tenant_id"] = tenant_id
fa_dict["company_id"] = company_id
@@ -263,11 +355,84 @@ class ItemService:
company_id: int,
) -> Item:
"""Update an item and optionally its nested data (multiple lines)"""
# Get existing item
db_item = ItemService.get_by_id(db, item_id, tenant_id, company_id)
if not db_item:
raise HTTPException(status_code=404, detail="Item not found")
# Validaciones con ErrorCollector
errors = ErrorCollector()
# Validar cada line item que se va a actualizar
if item_data.lines:
for idx, line_data in enumerate(item_data.lines):
# Si el line tiene ID, es actualización; si no, es creación
if hasattr(line_data, 'id') and line_data.id:
# Buscar el line item existente
existing_line = next(
(line for line in db_item.lines if line.id == line_data.id),
None
)
if existing_line:
# Convertir a LineItemUpdate para validar
line_update = LineItemUpdate(**line_data.model_dump())
validate_update(db, line_update, existing_line, errors)
else:
# Es un nuevo line item, validar como creación
line_create = LineItemCreate(**line_data.model_dump())
validate_create(db, line_create, tenant_id, company_id, errors)
# Validaciones adicionales específicas del negocio
# (Aplican tanto para crear como actualizar)
# Validar apóstrofes en número de parte
if line_data.part_number_id and "'" in str(line_data.part_number_id):
errors.add_error(
field=f"lines[{idx}].part_number_id",
message=f"Advertencia: El Número de Parte contiene apóstrofes y serán omitidos",
solution=None,
code="WARNING_APOSTROPHE"
)
# Validar tipo de partida
if hasattr(line_data, 'item_type'):
tipo_partida = line_data.item_type
if tipo_partida and tipo_partida not in ['N', 'S']:
errors.add_error(
field=f"lines[{idx}].item_type",
message=f"Tipo de partida debe ser 'N' (Normal) o 'S' (Subpartida), recibido: '{tipo_partida}'",
solution=None,
code="INVALID_ITEM_TYPE",
value=str(tipo_partida)
)
# Si es subpartida (S), debe tener partida principal
if tipo_partida == 'S':
if not hasattr(line_data, 'main_line_id') or not line_data.main_line_id:
errors.add_error(
field=f"lines[{idx}].main_line_id",
message="Las subpartidas (tipo 'S') deben tener una partida principal",
solution=None,
code="MISSING_MAIN_LINE"
)
# Validar que el line_number sea consecutivo (si se especifica)
if hasattr(line_data, 'line_number') and line_data.line_number:
expected_line = idx + 1
if line_data.line_number != expected_line:
errors.add_error(
field=f"lines[{idx}].line_number",
message=f"Número de línea esperado: {expected_line}, recibido: {line_data.line_number}",
solution=None,
code="INVALID_LINE_SEQUENCE",
value=str(line_data.line_number)
)
# Si hay errores, lanzar excepción ANTES de actualizar
errors.raise_if_errors("Error al actualizar el item")
try:
# Get existing item
db_item = ItemService.get_by_id(db, item_id, tenant_id, company_id)
if not db_item:
raise HTTPException(status_code=404, detail="Item not found")
# Extract lines data
lines_data = item_data.lines
@@ -343,7 +508,7 @@ class ItemService:
# Create FA data if provided
if fa_data is not None:
fa_dict = fa_data.model_dump(exclude_unset=True)
fa_dict = fa_data.model_dump(exclude_unset=True, exclude={"line_item_id"})
fa_dict["id"] = db_line.id # FA table uses same ID as line item
fa_dict["tenant_id"] = tenant_id
fa_dict["company_id"] = company_id