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

Refactor item validation and service logic

Browse Source 
- Consolidated item creation and update validation into a common function to reduce code duplication.
- Updated the `validate_create` and `validate_update` functions to utilize the new common validation logic.
- Introduced a new `common_validators.py` file for shared validation functions.
- Added a new `fractions.py` file to handle fraction-related logic and searches.
- Enhanced the `LineCustom` model to use an enumeration for `fraction_type`.
- Improved the `ItemService` class with methods for locking invoices and renumbering line items.
- Updated the `Sector` model to use a boolean type for the `authorized` field.
- Fixed import issues in the router by replacing the old `a24_router` with `sitar_router`.
This commit is contained in:
acazares
2026-02-04 23:29:05 -06:00
parent f38c13f851
commit 0440543f24
12 changed files with 692 additions and 748 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/api/v1/modules/a24/fa/fa_item_lines/models.py
View File

@@ -67,9 +67,6 @@ class FaLineItem(Base, TenantScopedMixin, TimestampMixin):
# Subitems
is_subitem: Mapped[Optional[bool]] = mapped_column(Boolean) # ESSUBPARTIDA
contains_subitems: Mapped[Optional[bool]] = mapped_column(Boolean) # CONTIENESUBP
includes_subitems: Mapped[Optional[bool]] = mapped_column(
Boolean
) # INCUYESUBPARTIDAS
subitem_number: Mapped[Optional[int]] = mapped_column(Integer) # SUBPARTIDA
# Special flags

12
backend/api/v1/modules/a76/general_catalogs/company/models.py
View File

@@ -39,7 +39,7 @@ class Company(Base, TimestampMixin):
# Programa
program: Mapped[Optional[str]] = mapped_column(String(7))
program_number: Mapped[Optional[str]] = mapped_column(String(40))
prosec: Mapped[Optional[int]] = mapped_column(SmallInteger)
prosec: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
prosec_authorization: Mapped[Optional[str]] = mapped_column(String(20))
# Sectores
@@ -61,9 +61,9 @@ class Company(Base, TimestampMixin):
# Configuración básica
logo: Mapped[Optional[str]] = mapped_column(String(255))
has_express_line: Mapped[Optional[str]] = mapped_column(String(2), default="N")
has_express_line: Mapped[Optional[bool]] = mapped_column(Boolean, default=False, server_default="false")
order_format_type: Mapped[Optional[str]] = mapped_column(String(19))
is_service_company: Mapped[Optional[bool]] = mapped_column(Boolean, default=False)
is_service_company: Mapped[Optional[bool]] = mapped_column(Boolean, default=False, server_default="false")
client_name: Mapped[Optional[str]] = mapped_column(String(300))
subassembly_mode: Mapped[Optional[str]] = mapped_column(String(7))
@@ -77,9 +77,11 @@ class Company(Base, TimestampMixin):
scaf_readonly: Mapped[Optional[int]] = mapped_column(SmallInteger)
parts_replacement: Mapped[Optional[int]] = mapped_column(SmallInteger)
activate_facmexame: Mapped[Optional[int]] = mapped_column(SmallInteger)
part_reference: Mapped[Optional[int]] = mapped_column(SmallInteger)
part_reference: Mapped[Optional[int]] = mapped_column(SmallInteger)
part_reference: Mapped[Optional[int]] = mapped_column(SmallInteger)
international_firm: Mapped[Optional[int]] = mapped_column(SmallInteger)
seventh_amendment: Mapped[Optional[bool]] = mapped_column(
Boolean
) # Septima enimenda (FinalContadorAElectronico)
# Configuraciones simples
ftp_key: Mapped[Optional[str]] = mapped_column(String(10))

51
backend/api/v1/modules/a76/invoices/common/common_validators.py
View File

@@ -1,3 +1,4 @@
from typing import Optional
from core.exceptions import ErrorCollector
from .. import models
from sqlalchemy.orm import Session
@@ -8,21 +9,53 @@ def invoice_exists(
invoice_number: str,
tenant_id: int,
company_id: int,
errors: ErrorCollector
) -> bool:
invoice_exists = (
db.query(models.InvoiceHeader.id)
errors: Optional[ErrorCollector],
):
invoice = (
db.query(models.InvoiceHeader)
.filter(
models.InvoiceHeader.invoice_number == invoice_number,
models.InvoiceHeader.tenant_id == tenant_id,
models.InvoiceHeader.company_id == company_id,
)
.first()
.first()
)
if invoice_exists:
errors.add_duplicate_error(
if invoice:
if errors:
errors.add_duplicate_error(
"invoice_number",
invoice_number,
f"Ya existe una factura con el número '{invoice_number}'",
)
)
return invoice
return None
def invoice_updated(
db: Session,
invoice_number: str,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
) -> bool:
is_updated = (
db.query(models.InvoiceHeader.is_updated)
.filter(
models.InvoiceHeader.invoice_number == invoice_number,
models.InvoiceHeader.tenant_id == tenant_id,
models.InvoiceHeader.company_id == company_id,
)
.scalar()
)
if is_updated:
errors.add_error(
field="invoice_number",
message=f"La factura con el número '{invoice_number}' ya ha sido actualizada y no se puede modificar.",
solution="Capturar otro número de Factura de Importación Temporal o Desactualizar la factura.",
code="INVOICE_UPDATED",
value=invoice_number,
)
return True
return False

25
backend/api/v1/modules/a76/items/common/common_validators.py Normal file
View File

@@ -0,0 +1,25 @@
from core.exceptions import ErrorCollector
from ..line_items import models
from sqlalchemy.orm import Session
def item_exists(
db: Session,
invoice_id: int,
item_line: int,
tenant_id: int,
company_id: int
):
item_exists = (
db.query(models.LineItem.id)
.filter(
models.LineItem.invoice_id == invoice_id,
models.LineItem.LineItem == item_line,
models.LineItem.tenant_id == tenant_id,
models.LineItem.company_id == company_id,
)
.scalar()
)
if item_exists:
return item_exists
return None

79
backend/api/v1/modules/a76/items/common/fractions.py Normal file
View File

@@ -0,0 +1,79 @@
import asyncio
from typing import Optional, Tuple
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.sitar.tlcs import TLCSService
def _search_historical_fraction() -> Tuple[Optional[str], float]:
"""Search for historical fraction data
TODO: Implement historical fraction search logic
This corresponds to BUSCA_FRACCION_HISTORICA in original Clarion code
Returns:
Tuple of (rate_im, adv_impo)
"""
# Placeholder for historical search
return None, 0.0
def search_fraction_preference(
country: str, fraction_type: str, company: Company, fraccion: str
) -> Tuple[Optional[str], float]:
"""Search fraction preference and return rate_im and adv_impo
Args:
country: Country code
fraction_type: Type of fraction (e.g., 'TLCS')
company: Company object with configuration
fraccion: Tariff fraction code to search
Returns:
Tuple of (rate_im, adv_impo) where:
- rate_im: Tax rate as string (e.g., "EXE", "5.0%")
- adv_impo: Numeric ad valorem rate
"""
rate_im = None
adv_impo = 0.0
country_group = "USA" if country == "MEX" else country
if fraction_type.upper() == "TLCS":
# Si tiene seventh_amendment configurado, consulta API externa
if company.seventh_amendment:
try:
# Get TLCS service instance
tlcs_service = TLCSService.get_instance()
# Fetch TLCS data using the service
tlcs_data = asyncio.run(
tlcs_service.search(
fraccion=fraccion, pais=country_group, limit=100
)
)
if not tlcs_data:
# No se encontraron registros, buscar en histórico
print(
f"No TLCS data found for fraction {fraccion[:8]}, searching historical"
)
return _search_historical_fraction()
else:
# Se encontraron registros, tomar el primero
first_record = tlcs_data[0]
rate_im = first_record.TASATXT
adv_impo = float(first_record.TASA1NUM or 0.0)
print(f"Found TLCS data: rate_im={rate_im}, adv_impo={adv_impo}")
except Exception as e:
print(f"Error fetching TLCS data from SITAR API: {e}")
return _search_historical_fraction()
else:
# Sin seventh_amendment, buscar en tabla local (base de datos)
# TODO: Implement local database search for TLCS
# This corresponds to the Access:GFracTLCSSifra.TryFetch logic in Clarion
print("Local TLCS search not implemented yet")
# Placeholder: search historical as fallback
return _search_historical_fraction()
return rate_im, adv_impo

410
backend/api/v1/modules/a76/items/imports/temporary/validators/common.py
View File

@@ -1,158 +1,282 @@
"""
Funciones helper compartidas para validaciones de items.
"""
from sqlalchemy.orm import Session
import logging
from typing import Optional, List, Tuple
from fastapi import HTTPException
from sqlalchemy import and_, or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, joinedload
from api.v1.modules.a76.invoices.common.common_validators import invoice_exists
from core.exceptions import ErrorCollector
from typing import Optional
from sqlalchemy import func
from ....common.fractions import search_fraction_preference
from ....common.common_validators import item_exists
from ....models import Item
from ....line_items.models import LineItem
from ....line_financials.models import LineFinancial
from ....line_quantities.models import LineQuantity
from ....line_customs.models import FractionType, LineCustom
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 api.v1.modules.a76.items.schemas import LineItemCreate
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
from api.v1.modules.a76.general_catalogs.packages.models import Package
from api.v1.modules.public.reference_data.countries.models import Country
from api.v1.modules.public.reference_data.sectors.models import Sector
from api.v1.modules.a76.general_catalogs.company.models import Company
def validate_catalog_reference(
def validate_common(
db: Session,
model_class,
id_value: Optional[int],
field_name: str,
line: LineItemCreate,
tenant_id: int,
company_id: Optional[int],
company_id: int,
errors: ErrorCollector,
error_message: str = None
) -> bool:
"""
Valida que una referencia a catálogo exista en la base de datos.
Args:
db: Sesión de base de datos
model_class: Clase del modelo SQLAlchemy a consultar
id_value: ID a validar
field_name: Nombre del campo para el error
tenant_id: ID del tenant
company_id: ID de la compañía (opcional)
errors: Colector de errores
error_message: Mensaje personalizado de error
Returns:
True si existe, False si no
"""
if not id_value:
return False
query = db.query(model_class).filter(
model_class.id == id_value,
model_class.tenant_id == tenant_id
line_number: int,
):
invoice: InvoiceHeader = invoice_exists(
db, line.invoice_id, tenant_id, company_id, errors
)
# Agregar filtro de company_id si el modelo lo tiene y se proporciona
if company_id and hasattr(model_class, 'company_id'):
query = query.filter(model_class.company_id == company_id)
exists = query.first() is not None
if not exists:
msg = error_message or f"El valor {id_value} no existe en el catálogo"
line_item: LineItem = item_exists(
db, line.invoice_id, line.line_item, tenant_id, company_id
)
fecha_factura = invoice.invoice_date if invoice else None
fraction = None
class_ = db.query(Class).filter(Class.id == line.class_id).first()
if not class_:
errors.add_error(
field=field_name,
message=msg,
solution="Selecciona un valor válido del catálogo",
code="NOT_FOUND"
field=f"line[{line_number}].class_id",
message="La clase especificada no existe.",
solution=["Darla de alta en el catalogo de clases."],
code="CLASS_NOT_FOUND",
)
return exists
def validate_positive_value(
value: Optional[float],
field_name: str,
errors: ErrorCollector,
required: bool = True,
allow_zero: bool = False
) -> bool:
"""
Valida que un valor numérico sea positivo.
Args:
value: Valor a validar
field_name: Nombre del campo para el error
errors: Colector de errores
required: Si el campo es obligatorio
allow_zero: Si se permite el valor cero
Returns:
True si es válido, False si no
"""
if value is None:
if required:
else:
if not line.unit_of_measure and not class_.unit_of_measure:
errors.add_error(
field=field_name,
message=f"El campo {field_name} es obligatorio",
solution="Proporciona un valor válido",
code="REQUIRED"
field=f"line[{line_number}].unit_of_measure",
message="La unidad de medida es obligatoria para la clase especificada.",
solution=["Proporciona una unidad de medida valida."],
code="UNIT_OF_MEASURE_REQUIRED",
)
return False
return True
if allow_zero and value == 0:
return True
if value <= 0:
if not line.customs.fraction:
if not line_item:
if not class_.fraction:
errors.add_error(
field=f"line[{line_number}].customs.fraction",
message="La fracción arancelaria es obligatoria para la clase especificada.",
solution=["Proporciona una fracción arancelaria valida."],
code="FRACTION_REQUIRED",
)
else:
fraction = class_.fraction
else:
if not line.customs.fraction:
if not class_.fraction:
errors.add_error(
field=f"line[{line_number}].customs.fraction",
message="La fracción arancelaria es obligatoria para la clase especificada.",
solution=["Proporciona una fracción arancelaria valida."],
code="FRACTION_REQUIRED",
)
else:
fraction = class_.fraction
else:
if line_item:
fraction = line.customs.fraction
if not line.description.description_spanish and not class_.description_es:
errors.add_error(
field=f"line[{line_number}].description.description_spanish",
message="La descripción en español es obligatoria para la clase especificada.",
solution=["Proporciona una descripción en español valida."],
code="DESCRIPTION_SPANISH_REQUIRED",
)
if not line.description.description_english and not class_.description_en:
errors.add_error(
field=f"line[{line_number}].description.description_english",
message="La descripción en inglés es obligatoria para la clase especificada.",
solution=["Proporciona una descripción en inglés valida."],
code="DESCRIPTION_ENGLISH_REQUIRED",
)
if line.quantity.quantity and line.quantity.quantity <= 0:
errors.add_error(
field=field_name,
message=f"El campo {field_name} debe ser mayor a cero",
solution="Proporciona un valor positivo",
code="INVALID_VALUE"
field=f"line[{line_number}].quantity.quantity",
message="La cantidad debe ser mayor a cero.",
solution=["Proporciona una cantidad valida."],
code="QUANTITY_MUST_BE_GREATER_THAN_ZERO",
)
return False
return True
def normalize_yes_no_value(value: Optional[str]) -> Optional[str]:
"""
Normaliza valores SI/NO a formato estándar.
Args:
value: Valor a normalizar (SI, NO, S, N)
Returns:
'SI' o 'NO', o None si el valor es None
"""
if not value:
return None
val = value.upper().strip()
if val in ['SI', 'S']:
return 'SI'
elif val in ['NO', 'N']:
return 'NO'
return value # Retornar original si no coincide
def validate_string_not_empty(
value: Optional[str],
field_name: str,
errors: ErrorCollector,
required: bool = True
) -> bool:
"""
Valida que un string no esté vacío.
Args:
value: Valor a validar
field_name: Nombre del campo para el error
errors: Colector de errores
required: Si el campo es obligatorio
Returns:
True si es válido, False si no
"""
if value is None or (isinstance(value, str) and not value.strip()):
if required:
errors.add_error(
field=field_name,
message=f"El campo {field_name} no puede estar vacío",
solution="Proporciona un valor válido",
code="REQUIRED"
if line.unit_of_measure:
um = (
db.query(func.count(UnitOfMeasure))
.filter(
UnitOfMeasure.code == line.unit_of_measure,
UnitOfMeasure.tenant_id == tenant_id,
UnitOfMeasure.company_id == company_id,
)
return False
return True
return True
.scalar()
)
if um == 0:
errors.add_error(
field=f"line[{line_number}].unit_of_measure",
message="La unidad de medida especificada no existe.",
solution=["Proporciona una unidad de medida valida."],
code="UNIT_OF_MEASURE_NOT_FOUND",
)
if line.quantity.package_id:
package = (
db.query(func.count(Package))
.filter(
Package.id == line.quantity.package_id,
Package.tenant_id == tenant_id,
Package.company_id == company_id,
)
.first()
)
if package == 0:
errors.add_error(
field=f"line[{line_number}].quantity.package_id",
message="El paquete especificado no existe.",
solution=["Proporciona un paquete valido."],
code="PACKAGE_NOT_FOUND",
)
if not line.quantity.package_quantity:
errors.add_error(
field=f"line[{line_number}].quantity.package_quantity",
message="La cantidad de paquetes es obligatoria cuando se proporciona el paquete.",
solution=["Proporciona una cantidad de paquetes valida."],
code="PACKAGE_QUANTITY_REQUIRED",
)
if line.quantity.package_quantity <= 0:
errors.add_error(
field=f"line[{line_number}].quantity.package_quantity",
message="La cantidad de paquetes debe ser mayor a cero.",
solution=["Proporciona una cantidad de paquetes valida."],
code="PACKAGE_QUANTITY_MUST_BE_GREATER_THAN_ZERO",
)
else:
if line.quantity.package_quantity and line.quantity.package_quantity > 0:
errors.add_error(
field=f"line[{line_number}].quantity.package_id",
message="El paquete es obligatorio cuando se proporciona la cantidad de paquetes.",
solution=["Proporciona un paquete valido."],
code="PACKAGE_ID_REQUIRED",
)
country = None
fraction_type = None
sector = None
if fraction:
fraction = line.customs.fraction if line.customs.fraction else fraction
country = line.customs.origin_country
if line_item:
country = (
line_item.customs.fraction if line_item.customs.origin_country else country
)
fraction_type = line.customs.fraction_type.upper()
if line_item:
fraction_type = (
line_item.customs.fraction_type
if line_item.customs.fraction_type
else fraction_type
)
sector = line.customs.sector
if line_item:
sector = line_item.customs.sector if line_item.customs.sector else sector
country_m3 = db.query(Country.m3_key).filter(Country.m3_key == country).scalar()
if not country_m3:
country_m3 = (
db.query(Country.m3_key).filter(Country.ame_key == country).scalar()
)
country = country_m3
if not country:
errors.add_error(
field=f"line[{line_number}].customs.origin_country",
message="El país de origen especificado no existe.",
solution=["Proporciona un país de origen valido."],
code="ORIGIN_COUNTRY_NOT_FOUND",
)
else:
if fraction_type.strip().upper() not in vars(FractionType).values():
valid_types = [
v
for k, v in vars(FractionType).items()
if not k.startswith("_") and isinstance(v, str)
]
errors.add_error(
field=f"line[{line_number}].customs.fraction_type",
message="El tipo de fracción especificado no es válido.",
solution=[
f"Proporciona un tipo de fracción válido. Valores permitidos: {', '.join(valid_types)}"
],
code="FRACTION_TYPE_INVALID",
value=fraction_type,
)
else:
if fraction_type.strip().upper() == FractionType.PROSEC and not sector:
errors.add_error(
field=f"line[{line_number}].customs.sector",
message="El sector es obligatorio cuando el tipo de fracción es 'PROSEC'.",
solution=["Proporciona un sector valido."],
code="SECTOR_REQUIRED_FOR_PROSEC",
)
elif fraction_type.strip().upper() != FractionType.PROSEC and sector:
errors.add_error(
field=f"line[{line_number}].customs.sector",
message="El sector solo es aplicable cuando el tipo de fracción es 'PROSEC'.",
solution=[
"Elimina el sector o cambia el tipo de fracción a 'PROSEC'."
],
code="SECTOR_ONLY_FOR_PROSEC",
)
elif fraction_type.strip().upper() == FractionType.PROSEC and sector:
sector_db: Sector = (
db.query(Sector)
.filter(
Class.sector_code == sector,
Class.tenant_id == tenant_id,
Class.company_id == company_id,
)
.scalar()
)
if sector_db:
errors.add_error(
field=f"line[{line_number}].customs.sector",
message="El sector especificado no existe.",
solution=["Proporciona un sector valido."],
code="SECTOR_NOT_FOUND",
)
else:
if not sector_db.authorized:
errors.add_error(
field=f"line[{line_number}].customs.sector",
message="El sector especificado no está autorizado.",
solution=["Proporciona un sector autorizado."],
code="SECTOR_NOT_AUTHORIZED",
)
company_db = db.query(Company).filter(Company.id == company_id).first()
if not company_db.prosec:
errors.add_error(
field=f"line[{line_number}].customs.sector",
message=" La empresa no cuenta con autorización PROSEC.",
solution=["Accese a los datos de la empresa y selecione la opción Pertenece al Programa de Promoción Sectorial y capture el número de permiso PROSEC."],
code="COMPANY_NOT_AUTHORIZED_FOR_PROSEC",
)
if fraction:
search_fraction_preference()

149
backend/api/v1/modules/a76/items/imports/temporary/validators/create.py
View File

@@ -1,135 +1,30 @@
"""
Validaciones para creación de items vía API.
"""
from sqlalchemy.orm import Session
import logging
from typing import Optional, List, Tuple
from fastapi import HTTPException
from sqlalchemy import and_, or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, joinedload
from api.v1.modules.a76.invoices.common.common_validators import invoice_exists
from core.exceptions import ErrorCollector
from api.v1.modules.a76.items.line_items.schemas import LineItemCreate
from .common import validate_string_not_empty, validate_positive_value
from ....line_items.models import LineItem
from ....line_financials.models import LineFinancial
from ....line_quantities.models import LineQuantity
from ....line_customs.models import LineCustom
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 .common import validate_common
def validate_create(
db: Session,
line: LineItemCreate,
line: Item,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
) -> None:
"""
Validaciones para crear LineItems vía API (actualmente en uso).
Args:
db: Sesión de base de datos
line: Datos del line item
tenant_id: ID del tenant
company_id: ID de la compañía
errors: Colector de errores
"""
# 1. Validar line_number
if not line.line_number:
errors.add_error(
field="line_number",
message="El número de línea es obligatorio",
solution="Proporciona un número de línea válido",
code="REQUIRED",
)
# 2. Validar class_id
if not line.class_id:
errors.add_error(
field="class_id",
message="Clase (ID) es obligatorio",
solution="Selecciona una clasificación válida del catálogo",
code="REQUIRED",
)
# 4. Validar unit_of_measure
if not line.unit_of_measure:
errors.add_error(
field="unit_of_measure",
message="U.M. es obligatorio",
solution="Proporciona una unidad de medida válida",
code="REQUIRED",
)
# 5. Validar quantity.quantity
if not line.quantity:
errors.add_error(
field="quantity",
message="Quantity es obligatorio",
solution="Proporciona una cantidad válida",
code="REQUIRED",
)
else:
# Validar con nombre amigable
if line.quantity.quantity is None or line.quantity.quantity <= 0:
errors.add_error(
field="quantity.quantity",
message="Quantity debe ser mayor a cero",
solution="Proporciona una cantidad válida",
code=(
"INVALID_VALUE"
if line.quantity.quantity is not None
else "REQUIRED"
),
)
# 6. Validar financial.unit_cost
if not line.financial:
errors.add_error(
field="financial",
message="Unit Cost es obligatorio",
solution="Proporciona el costo unitario del item",
code="REQUIRED",
)
else:
has_cost = (
line.financial.unit_cost_usd
or line.financial.unit_cost_mxn
or line.financial.unit_cost_capture
)
if not has_cost:
errors.add_error(
field="financial.unit_cost",
message="Unit Cost es obligatorio",
solution="Proporciona al menos un costo unitario (USD, MXN o captura)",
code="REQUIRED",
)
# 7. Validar description.description_spanish
if line.description:
if (
not line.description.description_spanish
or not line.description.description_spanish.strip()
):
errors.add_error(
field="description.description_spanish",
message="Description in Spanish es obligatorio",
solution="Proporciona una descripción del item en español",
code="REQUIRED",
)
else:
errors.add_error(
field="description.description_spanish",
message="Description in Spanish es obligatorio",
solution="Proporciona una descripción del item en español",
code="REQUIRED",
)
# 8. Validar customs.origin_country
if not line.customs or not line.customs.origin_country:
errors.add_error(
field="customs.origin_country",
message="País de Origen es obligatorio",
solution="Selecciona el país de origen del item",
code="REQUIRED"
)
# 9. Validar customs.fraction_type
if not line.customs or not line.customs.fraction_type:
errors.add_error(
field="customs.fraction_type",
message="Tipo de Tarifa es obligatorio",
solution="Selecciona el tipo de tarifa (GENERAL, PROSEC, ALADI, TLCS)",
code="REQUIRED"
)
line_number: int,
):
validate_common(db, line, tenant_id, company_id, errors, line_number)

260
backend/api/v1/modules/a76/items/imports/temporary/validators/update.py
View File

@@ -1,246 +1,30 @@
"""
Validaciones para actualización de items vía API.
"""
from sqlalchemy.orm import Session
import logging
from typing import Optional, List, Tuple
from fastapi import HTTPException
from sqlalchemy import and_, or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, joinedload
from api.v1.modules.a76.invoices.common.common_validators import invoice_exists
from core.exceptions import ErrorCollector
from api.v1.modules.a76.items.line_items.schemas import LineItemUpdate
from .common import validate_string_not_empty, validate_positive_value
from ....line_items.models import LineItem
from ....line_financials.models import LineFinancial
from ....line_quantities.models import LineQuantity
from ....line_customs.models import LineCustom
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 .common import validate_common
def validate_update(
db: Session,
line: LineItemUpdate,
line: Item,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
invoice_id: int = None,
) -> None:
"""
Validaciones para actualizar LineItems vía API.
Incluye todas las validaciones de negocio de Clarion.
Args:
db: Sesión de base de datos
line: Datos del line item a actualizar
tenant_id: ID del tenant
company_id: ID de la compañía
errors: Colector de errores
invoice_id: ID de la factura asociada (opcional, para validar subpartidas)
"""
# 1. Validar line_number si se proporciona
if line.line_number is not None and not line.line_number:
errors.add_error(
field="line_number",
message="El número de línea no puede estar vacío",
solution="Proporciona un número de línea válido",
code="REQUIRED",
)
# 2. Validar class_id si se proporciona
if line.class_id is not None and not line.class_id:
errors.add_error(
field="class_id",
message="Clase (ID) no puede estar vacío",
solution="Selecciona una clasificación válida del catálogo",
code="REQUIRED",
)
# 4. Validar unit_of_measure si se proporciona
if line.unit_of_measure is not None and not line.unit_of_measure:
errors.add_error(
field="unit_of_measure",
message="U.M. no puede estar vacío",
solution="Proporciona una unidad de medida válida",
code="REQUIRED",
)
# 5. Validar cantidad si se proporciona
if line.quantity:
# Si se proporciona el objeto quantity, validar que quantity.quantity sea válido
if line.quantity.quantity is not None:
if line.quantity.quantity <= 0:
errors.add_error(
field="quantity.quantity",
message="Quantity debe ser mayor a cero",
solution="Proporciona una cantidad válida",
code="INVALID_VALUE",
)
else:
# Si se proporciona quantity pero quantity.quantity es None, es requerido
errors.add_error(
field="quantity.quantity",
message="Quantity es obligatorio",
solution="Proporciona una cantidad mayor a 0",
code="REQUIRED",
)
# 6. Validar peso neto si se proporciona
if line.quantity and line.quantity.net_weight is not None:
if line.quantity.net_weight <= 0:
errors.add_error(
field="quantity.net_weight",
message="Net Weight debe ser mayor a cero",
solution="Proporciona un peso neto válido",
code="INVALID_VALUE",
)
# 7. Validar costo unitario si se proporciona financial (excepto subpartidas)
if line.financial:
is_subitem = line.fa_data and line.fa_data.is_subitem if line.fa_data else False
if not is_subitem:
has_cost = (
line.financial.unit_cost_usd
or line.financial.unit_cost_mxn
or line.financial.unit_cost_capture
)
if not has_cost:
errors.add_error(
field="financial.unit_cost",
message="Unit Cost es obligatorio",
solution="Proporciona al menos un costo unitario (USD, MXN o captura)",
code="REQUIRED",
)
# Validar que sean positivos
if line.financial.unit_cost_usd is not None:
if line.financial.unit_cost_usd <= 0:
errors.add_error(
field="financial.unit_cost_usd",
message="Unit Cost (USD) debe ser mayor a cero",
solution="Proporciona un costo unitario válido",
code="INVALID_VALUE",
)
if line.financial.unit_cost_mxn is not None:
if line.financial.unit_cost_mxn <= 0:
errors.add_error(
field="financial.unit_cost_mxn",
message="Unit Cost (MXN) debe ser mayor a cero",
solution="Proporciona un costo unitario válido",
code="INVALID_VALUE",
)
if line.financial.unit_cost_capture is not None:
if line.financial.unit_cost_capture <= 0:
errors.add_error(
field="financial.unit_cost_capture",
message="Unit Cost (Captura) debe ser mayor a cero",
solution="Proporciona un costo unitario válido",
code="INVALID_VALUE",
)
# 8. Validar datos aduanales si se proporcionan
if line.customs:
# Validar país de origen (OBLIGATORIO)
if line.customs.origin_country is not None:
if not line.customs.origin_country:
errors.add_error(
field="customs.origin_country",
message="País de Origen es obligatorio",
solution="Selecciona el país de origen del item",
code="REQUIRED"
)
# Validar tipo de tarifa (OBLIGATORIO)
if line.customs.fraction_type is not None:
if not line.customs.fraction_type:
errors.add_error(
field="customs.fraction_type",
message="Tipo de Tarifa es obligatorio",
solution="Selecciona el tipo de tarifa (GENERAL, PROSEC, ALADI, TLCS)",
code="REQUIRED"
)
# Validar preferencia arancelaria
if line.customs.preference is not None and not line.customs.preference:
errors.add_error(
field="customs.preference",
message="La preferencia arancelaria no puede estar vacía",
solution="Selecciona la preferencia arancelaria",
code="REQUIRED",
)
# Validar formato de pago de impuestos
if line.customs.tax_paid:
val_tax = line.customs.tax_paid.upper()
if val_tax not in ["SI", "NO", "S", "N"]:
errors.add_error(
field="customs.tax_paid",
message="El valor de pago de impuesto debe ser SI/NO o S/N",
solution="Proporciona un valor válido: SI, NO, S o N",
code="INVALID_VALUE",
)
# Validar forma de pago si existe
if line.customs.payment_form:
from api.v1.modules.a76.general_catalogs.forms_of_payment.models import (
PaymentForm,
)
payment = (
db.query(PaymentForm)
.filter(
PaymentForm.code == line.customs.payment_form,
PaymentForm.tenant_id == tenant_id,
)
.first()
)
if not payment:
errors.add_error(
field="customs.payment_form",
message=f"La forma de pago '{line.customs.payment_form}' no es válida",
solution="Selecciona una forma de pago válida del catálogo",
code="INVALID_VALUE",
)
# 9. Validar descripción en español (OBLIGATORIA)
if line.description and hasattr(line.description, 'description_spanish'):
if line.description.description_spanish is not None:
if not line.description.description_spanish.strip():
errors.add_error(
field="description.description_spanish",
message="Descripción en Español es obligatoria",
solution="Proporciona una descripción del item en español",
code="REQUIRED"
)
# 10. Validar subpartidas si se actualizan
if line.fa_data and line.fa_data.is_subitem:
# Es subpartida, debe tener partida principal
if not line.fa_data.main_line_id:
errors.add_error(
field="fa_data.main_line_id",
message="La subpartida debe tener asignada una partida principal",
solution="Selecciona la partida principal de esta subpartida",
code="REQUIRED",
)
elif invoice_id:
# Validar que la partida principal exista en la misma factura
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.items.models import Item
parent = (
db.query(LineItem)
.join(LineItem.item)
.filter(
LineItem.line_number == line.fa_data.main_line_id,
Item.invoice_id == invoice_id,
LineItem.company_id == company_id,
)
.first()
)
if not parent:
errors.add_error(
field="fa_data.main_line_id",
message=f"La partida principal {line.fa_data.main_line_id} no existe en esta factura",
solution="Verifica el número de la partida principal",
code="NOT_FOUND",
)
elif parent.fa_data and parent.fa_data.is_subitem:
errors.add_error(
field="fa_data.main_line_id",
message="La partida principal no puede ser otra subpartida",
solution="Selecciona una partida normal como principal",
code="INVALID_VALUE",
)
line_number: int,
):
validate_common(db, line, tenant_id, company_id, errors, line_number)

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

@@ -6,6 +6,13 @@ from core.database import Base
if TYPE_CHECKING:
from ..line_items.models import LineItem
class FractionType:
"""Enumeration for fraction types"""
GENERAL = "GENERAL"
PROSEC = "PROSEC"
ALADI = "ALADI"
TLCS = "TLCS"
class LineCustom(Base):
"""
@@ -22,7 +29,7 @@ class LineCustom(Base):
# Tariff/Customs Classifications
fraction: Mapped[Optional[str]] = mapped_column(String(10)) # FRACCION / FRACCIONIMPO / FRACCIONEXPO
fraction_type: Mapped[Optional[str]] = mapped_column(String(7)) # TIPOFRACCION / TIPOFRACCIONIMPO / TIPOFRACCIONEXPO
fraction_type: Mapped[Optional[FractionType]] = mapped_column(String(7)) # TIPOFRACCION / TIPOFRACCIONIMPO / TIPOFRACCIONEXPO
american_fraction: Mapped[Optional[str]] = mapped_column(String(16)) # FRACCIONAMERICANA
alternate_fraction: Mapped[Optional[str]] = mapped_column(String(10)) # FRACCIONALTERNA
reference_fraction: Mapped[Optional[str]] = mapped_column(String(10)) # FRACCIONREFERENCIA

429
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 api.v1.modules.a76.invoices.common.common_validators import (
invoice_exists,
invoice_updated,
)
from core.exceptions import ErrorCollector
from .imports.temporary.validators.create import validate_create
from .imports.temporary.validators.update import validate_update
@@ -32,8 +36,6 @@ 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__)
@@ -43,6 +45,101 @@ class ItemService:
Service for managing Items and related entities with tenant/company isolation
"""
@staticmethod
def _get_next_line_number(db: Session, invoice_id: int) -> int:
"""Calculate the next line_number for a given invoice based on database."""
from sqlalchemy import func
max_line = (
db.query(func.max(LineItem.line_number))
.join(Item, LineItem.item_id == Item.id)
.filter(Item.invoice_id == invoice_id)
.scalar()
)
return 1 if max_line is None else max_line + 1
@staticmethod
def _renumber_all_invoice_lines(db: Session, invoice_id: int) -> None:
"""Renumber all line_items for a given invoice to be consecutive (1, 2, 3, ...)."""
items = db.query(Item).filter(Item.invoice_id == invoice_id).all()
all_lines = [line for item in items for line in item.lines]
all_lines.sort(key=lambda x: x.line_number if x.line_number else 0)
for idx, line in enumerate(all_lines, start=1):
line.line_number = idx
@staticmethod
def _lock_invoice(
db: Session,
invoice_id: int,
tenant_id: int,
company_id: int,
errors: ErrorCollector,
) -> Optional[InvoiceHeader]:
"""Lock invoice to prevent concurrent modifications. Returns locked invoice or adds error."""
try:
invoice = (
db.query(InvoiceHeader)
.filter(
InvoiceHeader.id == invoice_id,
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
)
.with_for_update()
.first()
)
if not invoice:
errors.add_error(
field="invoice_id",
message="La factura no existe o no se pudo bloquear",
code="LOCK_FAILED",
value=str(invoice_id),
)
return invoice
except Exception as e:
logger.error(f"Error locking invoice {invoice_id}: {e}")
errors.add_error(
field="invoice_id",
message="Error al intentar bloquear la factura",
code="LOCK_ERROR",
)
return None
@staticmethod
def _create_line_nested_data(
db: Session, line: LineItem, line_data, tenant_id: int, company_id: int
) -> None:
"""Create all nested data for a line item."""
nested_models = [
(line_data.financial, LineFinancial),
(line_data.quantity, LineQuantity),
(line_data.customs, LineCustom),
(line_data.description, LineDescription),
(line_data.reference, LineReference),
]
for data, model_class in nested_models:
if data:
nested_dict = (
data.model_dump(exclude_unset=True)
if hasattr(data, "model_dump")
else data.model_dump()
)
nested_dict["item_line_id"] = line.id
db.add(model_class(**nested_dict))
# FA data uses line.id as primary key
if line_data.fa_data:
fa_dict = line_data.fa_data.model_dump(
exclude_unset=True, exclude={"line_item_id"}
)
fa_dict.update(
{"id": line.id, "tenant_id": tenant_id, "company_id": company_id}
)
db.add(FaLineItem(**fa_dict))
@staticmethod
def get_by_id(
db: Session, item_id: int, tenant_id: int, company_id: int
@@ -163,78 +260,61 @@ class ItemService:
errors = ErrorCollector()
# Validar que la factura exista y no esté actualizada (si viene invoice_id)
invoice = None
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 item_data.invoice_id:
errors.add_required_error(field="invoice_id")
return
if not invoice:
errors.add_error(
field="invoice_id",
message="La factura especificada no existe",
code="NOT_FOUND",
value=str(item_data.invoice_id),
)
if not invoice_exists(db, item_data.invoice_id, tenant_id, company_id, errors):
return
if not invoice_updated(
db, item_data.invoice_number, tenant_id, company_id, errors
):
return
# Lock invoice and pre-calculate line_numbers
if not ItemService._lock_invoice(
db, item_data.invoice_id, tenant_id, company_id, errors
):
errors.raise_if_errors("Error al crear el item")
line_numbers = []
if item_data.lines:
starting_line = ItemService._get_next_line_number(db, item_data.invoice_id)
line_numbers = [starting_line + i for i in range(len(item_data.lines))]
# Validar cada line item que se va a crear
if item_data.lines:
for idx, line_data in enumerate(item_data.lines):
line_number = line_numbers[idx] # Usar el line_number calculado
# Convertir a LineItemCreate para validar
line_create = LineItemCreate(**line_data.model_dump())
validate_create(db, line_create, tenant_id, company_id, errors)
validate_create(
db, line_create, tenant_id, company_id, errors, line_number
)
# Validaciones adicionales específicas del negocio
if not line_data.fa_data.is_subitem:
errors.add_required_error(
field=f"lines[{line_number}].fa_data.is_subitem"
)
return
if not line_data.fa_data.subitem_number:
errors.add_required_error(
field=f"lines[{line_number}].fa_data.subitem_number"
)
return
# 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",
field=f"lines[{line_number}].part_number",
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")
@@ -254,14 +334,6 @@ class ItemService:
# Create line items if provided
for idx, line_data in enumerate(lines_data):
# Extract nested data from line
financial_data = line_data.financial
quantity_data = line_data.quantity
customs_data = line_data.customs
description_data = line_data.description
reference_data = line_data.reference
fa_data = line_data.fa_data
line_dict = line_data.model_dump(
exclude={
"financial",
@@ -272,66 +344,35 @@ class ItemService:
"fa_data",
}
)
line_dict["item_id"] = db_item.id
line_dict["tenant_id"] = tenant_id
line_dict["company_id"] = company_id
line_dict.update(
{
"item_id": db_item.id,
"tenant_id": tenant_id,
"company_id": company_id,
"line_number": (
line_numbers[idx]
if line_numbers
else ItemService._get_next_line_number(
db, item_data.invoice_id
)
),
}
)
# Map schema field names to model field names
if "part_number_id" in line_dict:
line_dict["part_number"] = line_dict.pop("part_number_id")
if "component_part_number_id" in line_dict:
line_dict["component_part_number"] = line_dict.pop("component_part_number_id")
line_dict["part_number"] = line_dict.pop("part_number_id", None)
line_dict["component_part_number"] = line_dict.pop(
"component_part_number_id", None
)
# Create line item
db_line = LineItem(**line_dict)
db.add(db_line)
db.flush() # Get the line ID
db.flush()
# Create financial data if provided
if financial_data:
financial_dict = financial_data.model_dump()
financial_dict["item_line_id"] = db_line.id
db_financial = LineFinancial(**financial_dict)
db.add(db_financial)
# Create quantity data if provided
if quantity_data:
quantity_dict = quantity_data.model_dump()
quantity_dict["item_line_id"] = db_line.id
db_quantity = LineQuantity(**quantity_dict)
db.add(db_quantity)
# Create customs data if provided
if customs_data:
customs_dict = customs_data.model_dump()
customs_dict["item_line_id"] = db_line.id
db_customs = LineCustom(**customs_dict)
db.add(db_customs)
# Create description data if provided
if description_data:
description_dict = description_data.model_dump()
description_dict["item_line_id"] = db_line.id
db_description = LineDescription(**description_dict)
db.add(db_description)
# Create reference data if provided
if reference_data:
reference_dict = reference_data.model_dump()
reference_dict["item_line_id"] = db_line.id
db_reference = LineReference(**reference_dict)
db.add(db_reference)
# Create FA data if provided
if fa_data:
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
db_fa = FaLineItem(**fa_dict)
db.add(db_fa)
# Create all nested data
ItemService._create_line_nested_data(
db, db_line, line_data, tenant_id, company_id
)
db.commit()
db.refresh(db_item)
@@ -367,39 +408,25 @@ class ItemService:
# Validaciones con ErrorCollector
errors = ErrorCollector()
# Si se está actualizando el invoice_id, validar la factura
invoice = None
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()
)
# Lock invoice
invoice_id_to_lock = (
item_data.invoice_id if item_data.invoice_id else db_item.invoice_id
)
if not ItemService._lock_invoice(
db, invoice_id_to_lock, tenant_id, company_id, errors
):
errors.raise_if_errors("Error al actualizar el item")
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:
# Si no se está actualizando invoice_id, obtener la factura actual por invoice_id
from api.v1.modules.a76.invoices.models import InvoiceHeader
invoice = (
db.query(InvoiceHeader)
.filter(InvoiceHeader.id == db_item.invoice_id)
.first()
)
# Pre-calcular line_numbers para cada línea (en update, las líneas se renumeran desde 1)
line_numbers = []
if item_data.lines:
line_numbers = [i + 1 for i in range(len(item_data.lines))]
# Validar cada line item que se va a actualizar
if item_data.lines:
for idx, line_data in enumerate(item_data.lines):
line_number = line_numbers[idx] # Usar el line_number calculado
# 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
@@ -410,11 +437,15 @@ class ItemService:
if existing_line:
# Convertir a LineItemUpdate para validar
line_update = LineItemUpdate(**line_data.model_dump())
validate_update(db, line_update, tenant_id, company_id, errors)
validate_update(
db, line_update, tenant_id, company_id, errors, line_number
)
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)
validate_create(
db, line_create, tenant_id, company_id, errors, line_number
)
# Validaciones adicionales específicas del negocio
# (Aplican tanto para crear como actualizar)
@@ -422,7 +453,7 @@ class ItemService:
# 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",
field=f"lines[{line_number}].part_number",
message=f"Advertencia: El Número de Parte contiene apóstrofes y serán omitidos",
solution=None,
code="WARNING_APOSTROPHE",
@@ -433,7 +464,7 @@ class ItemService:
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",
field=f"lines[{line_number}].item_type",
message=f"Tipo de partida debe ser 'N' (Normal) o 'S' (Subpartida), recibido: '{tipo_partida}'",
solution=None,
code="INVALID_ITEM_TYPE",
@@ -447,24 +478,12 @@ class ItemService:
or not line_data.main_line_id
):
errors.add_error(
field=f"lines[{idx}].main_line_id",
field=f"lines[{line_number}].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")
@@ -486,15 +505,7 @@ class ItemService:
db.flush()
# Create new lines
for line_data in lines_data:
# Extract nested data from line
financial_data = line_data.financial
quantity_data = line_data.quantity
customs_data = line_data.customs
description_data = line_data.description
reference_data = line_data.reference
fa_data = line_data.fa_data
for idx, line_data in enumerate(lines_data):
line_dict = line_data.model_dump(
exclude={
"financial",
@@ -506,57 +517,32 @@ class ItemService:
},
exclude_unset=True,
)
line_dict["item_id"] = db_item.id
line_dict["tenant_id"] = tenant_id
line_dict["company_id"] = company_id
line_dict.update(
{
"item_id": db_item.id,
"tenant_id": tenant_id,
"company_id": company_id,
"line_number": idx + 1,
}
)
# Map schema field names to model field names
if "part_number_id" in line_dict:
line_dict["part_number"] = line_dict.pop("part_number_id")
if "component_part_number_id" in line_dict:
line_dict["component_part_number"] = line_dict.pop("component_part_number_id")
line_dict["part_number"] = line_dict.pop("part_number_id", None)
line_dict["component_part_number"] = line_dict.pop(
"component_part_number_id", None
)
db_line = LineItem(**line_dict)
db.add(db_line)
db.flush()
# Create nested data if provided
if financial_data is not None:
financial_dict = financial_data.model_dump(exclude_unset=True)
financial_dict["item_line_id"] = db_line.id
db.add(LineFinancial(**financial_dict))
# Create all nested data
ItemService._create_line_nested_data(
db, db_line, line_data, tenant_id, company_id
)
if quantity_data is not None:
quantity_dict = quantity_data.model_dump(exclude_unset=True)
quantity_dict["item_line_id"] = db_line.id
db.add(LineQuantity(**quantity_dict))
if customs_data is not None:
customs_dict = customs_data.model_dump(exclude_unset=True)
customs_dict["item_line_id"] = db_line.id
db.add(LineCustom(**customs_dict))
if description_data is not None:
description_dict = description_data.model_dump(
exclude_unset=True
)
description_dict["item_line_id"] = db_line.id
db.add(LineDescription(**description_dict))
if reference_data is not None:
reference_dict = reference_data.model_dump(exclude_unset=True)
reference_dict["item_line_id"] = db_line.id
db.add(LineReference(**reference_dict))
# Create FA data if provided
if fa_data is not None:
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
db.add(FaLineItem(**fa_dict))
# Renumber all lines for this invoice to ensure consecutive numbering
ItemService._renumber_all_invoice_lines(db, db_item.invoice_id)
db.commit()
db.refresh(db_item)
@@ -582,7 +568,20 @@ class ItemService:
if not db_item:
return False
invoice_id = db_item.invoice_id
errors = ErrorCollector()
# Lock the invoice
if not ItemService._lock_invoice(
db, invoice_id, tenant_id, company_id, errors
):
raise HTTPException(
status_code=404, detail="Invoice not found or could not be locked"
)
db.delete(db_item)
db.flush()
ItemService._renumber_all_invoice_lines(db, invoice_id)
db.commit()
return True

8
backend/api/v1/modules/public/reference_data/sectors/models.py
View File

@@ -1,5 +1,5 @@
from core.database import Base
from sqlalchemy import PrimaryKeyConstraint, SmallInteger, String
from sqlalchemy import Boolean, PrimaryKeyConstraint, SmallInteger, String
from sqlalchemy.orm import Mapped, mapped_column
@@ -15,9 +15,9 @@ class Sector(Base):
description: Mapped[str] = mapped_column(
String(150), nullable=False
) # descripción oficial (en español)
authorized: Mapped[SmallInteger] = mapped_column(
SmallInteger
) # 1 = autorizado, 0 = no autorizado
authorized: Mapped[bool] = mapped_column(
Boolean
) # True = autorizado, False = no autorizado
def __repr__(self):
return f"<Sector(key={self.key}, description={self.description}, authorized={self.authorized})>"