604 lines
23 KiB
Python
604 lines
23 KiB
Python
"""
|
|
Service layer for Items business logic
|
|
Handles CRUD operations for Item with complete one-to-one relationships:
|
|
Item -> LineItem -> LineFinancial
|
|
-> LineQuantity
|
|
-> LineCustoms
|
|
-> LineDescription
|
|
-> LineReference
|
|
-> FaLineItem (Fixed Assets - a24)
|
|
"""
|
|
|
|
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 api.v1.modules.a76.items.line_items.schemas import LineItemCreate, LineItemUpdate
|
|
|
|
from .schemas import ItemCreate, ItemUpdate
|
|
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
|
|
|
|
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))
|
|
.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
|
|
) -> Optional[Item]:
|
|
"""Get an item by ID with tenant/company validation"""
|
|
return (
|
|
db.query(Item)
|
|
.options(
|
|
joinedload(Item.lines).joinedload(LineItem.financial),
|
|
joinedload(Item.lines).joinedload(LineItem.quantity),
|
|
joinedload(Item.lines).joinedload(LineItem.customs),
|
|
joinedload(Item.lines).joinedload(LineItem.description),
|
|
joinedload(Item.lines).joinedload(LineItem.reference),
|
|
joinedload(Item.lines).joinedload(LineItem.class_info),
|
|
joinedload(Item.lines).joinedload(LineItem.unit_of_measure_info),
|
|
joinedload(Item.lines).joinedload(LineItem.fa_data),
|
|
)
|
|
.filter(
|
|
Item.id == item_id,
|
|
Item.tenant_id == tenant_id,
|
|
Item.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[Item], int]:
|
|
"""Get all items for a tenant/company with pagination and optional filters"""
|
|
query = (
|
|
db.query(Item)
|
|
.options(
|
|
joinedload(Item.lines).joinedload(LineItem.financial),
|
|
joinedload(Item.lines).joinedload(LineItem.quantity),
|
|
joinedload(Item.lines).joinedload(LineItem.customs),
|
|
joinedload(Item.lines).joinedload(LineItem.description),
|
|
joinedload(Item.lines).joinedload(LineItem.reference),
|
|
joinedload(Item.lines).joinedload(LineItem.class_info),
|
|
joinedload(Item.lines).joinedload(LineItem.unit_of_measure_info),
|
|
joinedload(Item.lines).joinedload(LineItem.fa_data),
|
|
)
|
|
.filter(
|
|
Item.tenant_id == tenant_id,
|
|
Item.company_id == company_id,
|
|
)
|
|
)
|
|
|
|
# Apply filters if provided
|
|
if filters:
|
|
if filters.get("invoice_id"):
|
|
query = query.filter(Item.invoice_id == filters["invoice_id"])
|
|
if filters.get("item_type"):
|
|
query = query.filter(Item.item_type == filters["item_type"])
|
|
if filters.get("system_origin"):
|
|
query = query.filter(Item.system_origin == filters["system_origin"])
|
|
if filters.get("search"):
|
|
search_term = f"%{filters['search']}%"
|
|
query = query.filter(
|
|
or_(
|
|
Item.invoice_id.ilike(search_term),
|
|
Item.reference_number.ilike(search_term),
|
|
Item.order.ilike(search_term),
|
|
Item.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[Item], int]:
|
|
"""Get all items for a specific invoice"""
|
|
query = (
|
|
db.query(Item)
|
|
.options(
|
|
joinedload(Item.lines).joinedload(LineItem.financial),
|
|
joinedload(Item.lines).joinedload(LineItem.quantity),
|
|
joinedload(Item.lines).joinedload(LineItem.customs),
|
|
joinedload(Item.lines).joinedload(LineItem.description),
|
|
joinedload(Item.lines).joinedload(LineItem.reference),
|
|
joinedload(Item.lines).joinedload(LineItem.fa_data),
|
|
)
|
|
.filter(
|
|
Item.invoice_id == invoice_id,
|
|
Item.tenant_id == tenant_id,
|
|
Item.company_id == company_id,
|
|
)
|
|
)
|
|
|
|
total = query.count()
|
|
items = query.offset(skip).limit(limit).all()
|
|
return items, total
|
|
|
|
@staticmethod
|
|
def create(
|
|
db: Session,
|
|
item_data: ItemCreate,
|
|
tenant_id: int,
|
|
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 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 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
|
|
|
|
validate_create(
|
|
db,
|
|
line_data, # Schema Pydantic completo
|
|
item_data.invoice_id, # invoice_id
|
|
tenant_id,
|
|
company_id,
|
|
errors,
|
|
line_number,
|
|
)
|
|
|
|
# Validaciones adicionales específicas del negocio
|
|
if line_data.fa_data and line_data.fa_data.is_subitem is None:
|
|
errors.add_required_error(
|
|
field=f"lines[{line_number}].fa_data.is_subitem"
|
|
)
|
|
|
|
if line_data.fa_data and line_data.fa_data.subitem_number is None:
|
|
errors.add_required_error(
|
|
field=f"lines[{line_number}].fa_data.subitem_number"
|
|
)
|
|
|
|
# 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[{line_number}].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:
|
|
# Extract lines data
|
|
lines_data = item_data.lines or []
|
|
item_dict = item_data.model_dump(exclude={"lines"})
|
|
|
|
# Add tenant and company
|
|
item_dict["tenant_id"] = tenant_id
|
|
item_dict["company_id"] = company_id
|
|
|
|
# Create the item
|
|
db_item = Item(**item_dict)
|
|
db.add(db_item)
|
|
db.flush() # Get the item ID
|
|
|
|
# Create line items if provided
|
|
for idx, line_data in enumerate(lines_data):
|
|
line_dict = line_data.model_dump(
|
|
exclude={
|
|
"financial",
|
|
"quantity",
|
|
"customs",
|
|
"description",
|
|
"reference",
|
|
"fa_data",
|
|
}
|
|
)
|
|
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
|
|
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 all nested data
|
|
ItemService._create_line_nested_data(
|
|
db, db_line, line_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="Item creation failed - integrity constraint violated",
|
|
)
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Unexpected error creating item: {e}")
|
|
raise HTTPException(status_code=500, detail="Error creating item")
|
|
|
|
@staticmethod
|
|
def update(
|
|
db: Session,
|
|
item_id: int,
|
|
item_data: ItemUpdate,
|
|
tenant_id: int,
|
|
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()
|
|
|
|
# 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")
|
|
|
|
# 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
|
|
existing_line = next(
|
|
(line for line in db_item.lines if line.id == line_data.id),
|
|
None,
|
|
)
|
|
if existing_line:
|
|
# Validar update con línea existente
|
|
validate_update(
|
|
db,
|
|
line_data, # Schema de update
|
|
existing_line, # LineItem existente en DB
|
|
invoice_id_to_lock, # invoice_id
|
|
tenant_id,
|
|
company_id,
|
|
errors,
|
|
line_number,
|
|
)
|
|
else:
|
|
# Es un nuevo line item, validar como creación
|
|
validate_create(
|
|
db,
|
|
line_data, # Schema Pydantic completo
|
|
invoice_id_to_lock, # invoice_id
|
|
tenant_id,
|
|
company_id,
|
|
errors,
|
|
line_number,
|
|
)
|
|
|
|
# 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[{line_number}].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(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[{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",
|
|
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[{line_number}].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:
|
|
|
|
# Extract lines data
|
|
lines_data = item_data.lines
|
|
item_dict = item_data.model_dump(exclude={"lines"}, exclude_unset=True)
|
|
|
|
# Update item fields
|
|
for key, value in item_dict.items():
|
|
setattr(db_item, key, value)
|
|
|
|
# Update lines if provided (replace all lines)
|
|
if lines_data is not None:
|
|
# Delete existing lines (cascade will handle nested data)
|
|
for existing_line in db_item.lines:
|
|
db.delete(existing_line)
|
|
db.flush()
|
|
|
|
# Create new lines
|
|
for idx, line_data in enumerate(lines_data):
|
|
line_dict = line_data.model_dump(
|
|
exclude={
|
|
"financial",
|
|
"quantity",
|
|
"customs",
|
|
"description",
|
|
"reference",
|
|
"fa_data",
|
|
},
|
|
exclude_unset=True,
|
|
)
|
|
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
|
|
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 all nested data
|
|
ItemService._create_line_nested_data(
|
|
db, db_line, line_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")
|