c040679fc5
- Introduced nested interfaces for line items including customs, financials, quantities, descriptions, and references. - Updated Item interface to include lines as an array of LineItem. - Modified invoice top fields to handle pedimento ID and auto-assign values from the invoice. - Enhanced item configuration component to manage line item properties and descriptions. - Updated main data, packages section, summary section, and other components to bind new line item properties. - Implemented normalization of numeric values when editing items to ensure consistent data types. - Adjusted save invoice logic to accommodate new line item structure and compliance data.
376 lines
14 KiB
Python
376 lines
14 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
|
|
"""
|
|
|
|
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.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 .models import Item
|
|
|
|
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),
|
|
)
|
|
.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),
|
|
)
|
|
.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),
|
|
)
|
|
.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)"""
|
|
try:
|
|
# Extract lines data
|
|
lines_data = item_data.lines or []
|
|
item_dict = item_data.model_dump(exclude={"lines"})
|
|
|
|
# DEBUG: Log incoming data
|
|
print(f"\n🔍 DEBUG CREATE ITEM:")
|
|
print(f" Item data: {item_dict}")
|
|
print(f" Lines count: {len(lines_data)}")
|
|
print(f" Tenant ID: {tenant_id}, Company ID: {company_id}")
|
|
|
|
# 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
|
|
|
|
print(f" ✅ Item created with ID: {db_item.id}")
|
|
|
|
# Create line items if provided
|
|
for idx, line_data in enumerate(lines_data):
|
|
print(f"\n 📝 Processing line {idx + 1}/{len(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
|
|
|
|
print(f" Line data: {line_data.model_dump()}")
|
|
print(f" Has financial: {financial_data is not None}")
|
|
print(f" Has quantity: {quantity_data is not None}")
|
|
print(f" Has customs: {customs_data is not None}")
|
|
print(f" Has description: {description_data is not None}")
|
|
print(f" Has reference: {reference_data is not None}")
|
|
|
|
line_dict = line_data.model_dump(
|
|
exclude={"financial", "quantity",
|
|
"customs", "description", "reference"}
|
|
)
|
|
line_dict["item_id"] = db_item.id
|
|
line_dict["tenant_id"] = tenant_id
|
|
line_dict["company_id"] = company_id
|
|
|
|
# Create line item
|
|
db_line = LineItem(**line_dict)
|
|
db.add(db_line)
|
|
db.flush() # Get the line ID
|
|
print(f" ✅ Line created with ID: {db_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)
|
|
print(f" ✅ Financial data added")
|
|
|
|
# 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)
|
|
print(f" ✅ Quantity data added")
|
|
|
|
# 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)
|
|
print(f" ✅ Customs data added")
|
|
|
|
# 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)
|
|
print(f" ✅ Description data added")
|
|
|
|
# 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)
|
|
print(f" ✅ Reference data added")
|
|
|
|
print(f"\n 💾 Committing transaction...")
|
|
db.commit()
|
|
db.refresh(db_item)
|
|
print(f" ✅ Transaction committed successfully!")
|
|
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)"""
|
|
try:
|
|
# Get existing item
|
|
db_item = ItemService.get_by_id(db, item_id, tenant_id, company_id)
|
|
if not db_item:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
|
|
# Extract lines data
|
|
lines_data = item_data.lines
|
|
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
|
|
|
|
line_dict = line_data.model_dump(
|
|
exclude={"financial", "quantity",
|
|
"customs", "description", "reference"},
|
|
exclude_unset=True
|
|
)
|
|
line_dict["item_id"] = db_item.id
|
|
line_dict["tenant_id"] = tenant_id
|
|
line_dict["company_id"] = company_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))
|
|
|
|
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")
|