acazares
b94d0ac7fb
- Renamed LineItem interface to Item and adjusted properties accordingly. - Updated CreateItemData and UpdateItemData interfaces to reflect new structure. - Modified components to use the new Item interface, removing nested lines. - Adjusted data binding in item configuration, main data, and other related components. - Simplified item creation and editing logic by removing unnecessary nesting. - Ensured all references to line items are updated to reflect the new structure.
531 lines
18 KiB
Python
531 lines
18 KiB
Python
"""
|
|
Service layer for Items business logic
|
|
Handles CRUD operations for LineItem with complete one-to-one relationships:
|
|
LineItem -> LineFinancial
|
|
-> LineQuantity
|
|
-> LineCustoms
|
|
-> LineDescription
|
|
-> LineReference
|
|
-> FaLineItem (Fixed Assets - a24)
|
|
|
|
After refactoring: LineItem is the main entity, representing a single line item in an invoice.
|
|
There is no intermediate Item entity anymore. Each LineItem belongs directly to an InvoiceHeader.
|
|
"""
|
|
|
|
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_by_id,
|
|
invoice_updated,
|
|
)
|
|
from core.exceptions import ErrorCollector
|
|
from .imports.temporary.validators.create import validate_create
|
|
from .imports.temporary.validators.update import validate_update
|
|
|
|
from .schemas import LineItemCreate, LineItemUpdate
|
|
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 LineItem
|
|
from api.v1.modules.a76.invoices.models import InvoiceHeader
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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))
|
|
.filter(LineItem.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(LineItem)
|
|
.filter(LineItem.invoice_id == invoice_id)
|
|
.order_by(LineItem.line_number)
|
|
.all()
|
|
)
|
|
|
|
for idx, item in enumerate(items, start=1):
|
|
item.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
|
|
) -> Optional[LineItem]:
|
|
"""Get an item by ID with tenant/company validation"""
|
|
return (
|
|
db.query(LineItem)
|
|
.options(
|
|
joinedload(LineItem.financial),
|
|
joinedload(LineItem.quantity),
|
|
joinedload(LineItem.customs),
|
|
joinedload(LineItem.description),
|
|
joinedload(LineItem.reference),
|
|
joinedload(LineItem.class_info),
|
|
joinedload(LineItem.unit_of_measure_info),
|
|
joinedload(LineItem.fa_data),
|
|
)
|
|
.filter(
|
|
LineItem.id == item_id,
|
|
LineItem.tenant_id == tenant_id,
|
|
LineItem.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
def get_all(
|
|
db: Session,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
filters: Optional[dict] = None,
|
|
) -> Tuple[List[LineItem], int]:
|
|
"""Get all items for a tenant/company with pagination and optional filters"""
|
|
query = (
|
|
db.query(LineItem)
|
|
.options(
|
|
joinedload(LineItem.financial),
|
|
joinedload(LineItem.quantity),
|
|
joinedload(LineItem.customs),
|
|
joinedload(LineItem.description),
|
|
joinedload(LineItem.reference),
|
|
joinedload(LineItem.class_info),
|
|
joinedload(LineItem.unit_of_measure_info),
|
|
joinedload(LineItem.fa_data),
|
|
)
|
|
.filter(
|
|
LineItem.tenant_id == tenant_id,
|
|
LineItem.company_id == company_id,
|
|
)
|
|
)
|
|
|
|
# Apply filters if provided
|
|
if filters:
|
|
if filters.get("invoice_id"):
|
|
query = query.filter(LineItem.invoice_id == filters["invoice_id"])
|
|
if filters.get("item_type"):
|
|
query = query.filter(LineItem.item_type == filters["item_type"])
|
|
if filters.get("system_origin"):
|
|
query = query.filter(LineItem.system_origin == filters["system_origin"])
|
|
if filters.get("search"):
|
|
search_term = f"%{filters['search']}%"
|
|
query = query.filter(
|
|
or_(
|
|
LineItem.invoice_id.ilike(search_term),
|
|
LineItem.reference_number.ilike(search_term),
|
|
LineItem.order.ilike(search_term),
|
|
LineItem.guide_number.ilike(search_term),
|
|
)
|
|
)
|
|
|
|
total = query.count()
|
|
items = query.offset(skip).limit(limit).all()
|
|
return items, total
|
|
|
|
@staticmethod
|
|
def get_by_invoice(
|
|
db: Session,
|
|
invoice_id: int,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> Tuple[List[LineItem], int]:
|
|
"""Get all items for a specific invoice"""
|
|
query = (
|
|
db.query(LineItem)
|
|
.options(
|
|
joinedload(LineItem.financial),
|
|
joinedload(LineItem.quantity),
|
|
joinedload(LineItem.customs),
|
|
joinedload(LineItem.description),
|
|
joinedload(LineItem.reference),
|
|
joinedload(LineItem.fa_data),
|
|
)
|
|
.filter(
|
|
LineItem.invoice_id == invoice_id,
|
|
LineItem.tenant_id == tenant_id,
|
|
LineItem.company_id == company_id,
|
|
)
|
|
)
|
|
|
|
total = query.count()
|
|
items = query.offset(skip).limit(limit).all()
|
|
return items, total
|
|
|
|
@staticmethod
|
|
def create(
|
|
db: Session,
|
|
item_data: LineItemCreate,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
) -> LineItem:
|
|
"""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 not item_data.invoice_id:
|
|
errors.add_required_error(field="invoice_id")
|
|
errors.raise_if_errors("Error al crear el item")
|
|
|
|
if not invoice_exists_by_id(
|
|
db, item_data.invoice_id, tenant_id, company_id, errors
|
|
):
|
|
errors.raise_if_errors("Error al crear el item")
|
|
if not invoice_updated(db, item_data.invoice_id, tenant_id, company_id, errors):
|
|
errors.raise_if_errors("Error al crear el item")
|
|
|
|
# Lock invoice and calculate line number
|
|
if not ItemService._lock_invoice(
|
|
db, item_data.invoice_id, tenant_id, company_id, errors
|
|
):
|
|
errors.raise_if_errors("Error al crear el item")
|
|
|
|
# Calculate the next line number for this single item
|
|
line_number = ItemService._get_next_line_number(db, item_data.invoice_id)
|
|
|
|
# Validar el item
|
|
validate_create(
|
|
db,
|
|
item_data, # Schema Pydantic completo
|
|
tenant_id,
|
|
company_id,
|
|
errors,
|
|
line_number,
|
|
)
|
|
|
|
# Validaciones adicionales específicas del negocio
|
|
if item_data.fa_data and item_data.fa_data.is_subitem is None:
|
|
errors.add_required_error(field=f"fa_data.is_subitem")
|
|
|
|
if item_data.fa_data and item_data.fa_data.subitem_number is None:
|
|
errors.add_required_error(field=f"fa_data.subitem_number")
|
|
|
|
# Validar apóstrofes en número de parte
|
|
if item_data.part_number_id and "'" in str(item_data.part_number_id):
|
|
errors.add_error(
|
|
field=f"part_number",
|
|
message=f"Advertencia: El Número de Parte contiene apóstrofes y serán omitidos",
|
|
code="WARNING_APOSTROPHE",
|
|
)
|
|
|
|
# Si hay errores, lanzar excepción ANTES de intentar crear
|
|
errors.raise_if_errors("Error al crear el item")
|
|
|
|
try:
|
|
# Prepare item data
|
|
item_dict = item_data.model_dump(
|
|
exclude={
|
|
"financial",
|
|
"quantity",
|
|
"customs",
|
|
"description",
|
|
"reference",
|
|
"fa_data",
|
|
}
|
|
)
|
|
|
|
# Add tenant, company and line number
|
|
item_dict.update(
|
|
{
|
|
"tenant_id": tenant_id,
|
|
"company_id": company_id,
|
|
"line_number": line_number,
|
|
}
|
|
)
|
|
|
|
# Create the item
|
|
db_item = LineItem(**item_dict)
|
|
db.add(db_item)
|
|
db.flush() # Get the item ID
|
|
|
|
# Create all nested data
|
|
ItemService._create_line_nested_data(
|
|
db, db_item, item_data, tenant_id, company_id
|
|
)
|
|
|
|
db.commit()
|
|
db.refresh(db_item)
|
|
return db_item
|
|
|
|
except IntegrityError as e:
|
|
db.rollback()
|
|
logger.error(f"Error creating item: {e}")
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="LineItem creation failed - integrity constraint violated",
|
|
)
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Unexpected error creating LineItem: {e}")
|
|
raise HTTPException(status_code=500, detail="Error creating LineItem")
|
|
|
|
@staticmethod
|
|
def update(
|
|
db: Session,
|
|
item_id: int,
|
|
item_data: LineItemUpdate,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
) -> LineItem:
|
|
"""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="LineItem not found")
|
|
|
|
# Validaciones con ErrorCollector
|
|
errors = ErrorCollector()
|
|
|
|
# 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")
|
|
|
|
# Validar el item que se va a actualizar
|
|
validate_update(
|
|
db,
|
|
item_data, # Schema de update
|
|
db_item, # LineItem existente en DB
|
|
tenant_id,
|
|
company_id,
|
|
errors,
|
|
db_item.line_number,
|
|
)
|
|
|
|
# Validaciones adicionales específicas del negocio
|
|
|
|
# Validar apóstrofes en número de parte
|
|
if item_data.part_number_id and "'" in str(item_data.part_number_id):
|
|
errors.add_error(
|
|
field=f"part_number",
|
|
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(item_data, "item_type") and item_data.item_type:
|
|
tipo_partida = item_data.item_type
|
|
if tipo_partida and tipo_partida not in ["N", "S"]:
|
|
errors.add_error(
|
|
field=f"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(item_data, "main_line_id") or not item_data.main_line_id:
|
|
errors.add_error(
|
|
field=f"main_line_id",
|
|
message="Las subpartidas (tipo 'S') deben tener una partida principal",
|
|
solution=None,
|
|
code="MISSING_MAIN_LINE",
|
|
)
|
|
|
|
# Si hay errores, lanzar excepción ANTES de actualizar
|
|
errors.raise_if_errors("Error al actualizar el item")
|
|
|
|
try:
|
|
# Get item data excluding nested objects
|
|
item_dict = item_data.model_dump(
|
|
exclude={
|
|
"financial",
|
|
"quantity",
|
|
"customs",
|
|
"description",
|
|
"reference",
|
|
"fa_data",
|
|
},
|
|
exclude_unset=True,
|
|
)
|
|
|
|
# Map schema field names to model field names
|
|
if "part_number_id" in item_dict:
|
|
item_dict["part_number"] = item_dict.pop("part_number_id")
|
|
if "component_part_number_id" in item_dict:
|
|
item_dict["component_part_number"] = item_dict.pop(
|
|
"component_part_number_id"
|
|
)
|
|
|
|
# Update item fields
|
|
for key, value in item_dict.items():
|
|
setattr(db_item, key, value)
|
|
|
|
# Delete existing nested data
|
|
db.query(LineFinancial).filter(
|
|
LineFinancial.item_line_id == db_item.id
|
|
).delete()
|
|
db.query(LineQuantity).filter(
|
|
LineQuantity.item_line_id == db_item.id
|
|
).delete()
|
|
db.query(LineCustom).filter(LineCustom.item_line_id == db_item.id).delete()
|
|
db.query(LineDescription).filter(
|
|
LineDescription.item_line_id == db_item.id
|
|
).delete()
|
|
db.query(LineReference).filter(
|
|
LineReference.item_line_id == db_item.id
|
|
).delete()
|
|
db.query(FaLineItem).filter(FaLineItem.id == db_item.id).delete()
|
|
db.flush()
|
|
|
|
# Create new nested data
|
|
ItemService._create_line_nested_data(
|
|
db, db_item, item_data, tenant_id, company_id
|
|
)
|
|
|
|
# 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)
|
|
return db_item
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Unexpected error updating item: {e}")
|
|
raise HTTPException(status_code=500, detail="Error updating item")
|
|
|
|
@staticmethod
|
|
def delete(
|
|
db: Session,
|
|
item_id: int,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
) -> bool:
|
|
"""Delete an item and all its related data (cascade delete)"""
|
|
try:
|
|
db_item = ItemService.get_by_id(db, item_id, tenant_id, company_id)
|
|
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
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error deleting item: {e}")
|
|
raise HTTPException(status_code=500, detail="Error deleting item")
|