572 lines
24 KiB
Python
572 lines
24 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 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
|
|
from api.v1.modules.a76.general_catalogs.company.models import Company
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ItemService:
|
|
"""
|
|
Service for managing Items and related entities with tenant/company isolation
|
|
"""
|
|
|
|
@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_number.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)
|
|
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 invoice:
|
|
errors.add_error(
|
|
field="invoice_id",
|
|
message="La factura especificada no existe",
|
|
code="NOT_FOUND",
|
|
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 []
|
|
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):
|
|
# 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",
|
|
"quantity",
|
|
"customs",
|
|
"description",
|
|
"reference",
|
|
"fa_data",
|
|
}
|
|
)
|
|
line_dict["item_id"] = db_item.id
|
|
line_dict["tenant_id"] = tenant_id
|
|
line_dict["company_id"] = company_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")
|
|
|
|
# Create line item
|
|
db_line = LineItem(**line_dict)
|
|
db.add(db_line)
|
|
db.flush() # Get the line ID
|
|
|
|
# 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)
|
|
|
|
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()
|
|
|
|
# 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()
|
|
|
|
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()
|
|
|
|
# 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, tenant_id, company_id, 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:
|
|
|
|
# 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 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
|
|
|
|
line_dict = line_data.model_dump(
|
|
exclude={
|
|
"financial",
|
|
"quantity",
|
|
"customs",
|
|
"description",
|
|
"reference",
|
|
"fa_data",
|
|
},
|
|
exclude_unset=True,
|
|
)
|
|
line_dict["item_id"] = db_item.id
|
|
line_dict["tenant_id"] = tenant_id
|
|
line_dict["company_id"] = company_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")
|
|
|
|
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))
|
|
|
|
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))
|
|
|
|
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
|
|
|
|
db.delete(db_item)
|
|
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")
|