304f7b07d4
- Implemented the items tab form in Svelte for editing invoices, displaying item quantities and import values. - Created saveInvoice function to handle both creation and updating of invoices, including validation of required fields and building a unified payload for API requests. - Added support for compliance, financials, and logistics data in the invoice payload.
801 lines
27 KiB
Python
801 lines
27 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"})
|
|
|
|
# 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 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"}
|
|
)
|
|
line_dict["item_id"] = db_item.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)
|
|
|
|
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)"""
|
|
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
|
|
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")
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def create_item(self, item_data: ItemCreate) -> Item:
|
|
"""
|
|
Create a new item with its nested line item (one-to-one)
|
|
Also creates the line's financial and quantity data
|
|
"""
|
|
try:
|
|
# Extract line data before creating the item
|
|
line_data = item_data.line
|
|
item_dict = item_data.model_dump(exclude={'line'})
|
|
|
|
# Create the item
|
|
db_item = Item(**item_dict)
|
|
self.db.add(db_item)
|
|
self.db.flush() # Get the item ID without committing
|
|
|
|
# Create line item with nested data if provided
|
|
if line_data:
|
|
self._create_line_item(db_item.id, line_data)
|
|
|
|
self.db.commit()
|
|
self.db.refresh(db_item)
|
|
return db_item
|
|
|
|
except IntegrityError as e:
|
|
self.db.rollback()
|
|
logger.error(f"Integrity error creating item: {e}")
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Item creation failed due to data integrity constraint"
|
|
)
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Unexpected error creating item: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Error creating item: {str(e)}"
|
|
)
|
|
|
|
def _create_line_item(self, item_id: int, line_data: LineItemCreate) -> LineItem:
|
|
"""
|
|
Create a line item with all its nested data (financial, quantity, customs, description, reference)
|
|
"""
|
|
# Extract nested data
|
|
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'
|
|
})
|
|
|
|
# Create line item
|
|
db_line = LineItem(item_id=item_id, **line_dict)
|
|
self.db.add(db_line)
|
|
self.db.flush() # Get the line ID
|
|
|
|
# Create financial data if provided
|
|
if financial_data:
|
|
db_financial = LineFinancial(
|
|
item_line_id=db_line.id,
|
|
**financial_data.model_dump()
|
|
)
|
|
self.db.add(db_financial)
|
|
|
|
# Create quantity data if provided
|
|
if quantity_data:
|
|
db_quantity = LineQuantity(
|
|
item_line_id=db_line.id,
|
|
**quantity_data.model_dump()
|
|
)
|
|
self.db.add(db_quantity)
|
|
|
|
# Create customs data if provided
|
|
if customs_data:
|
|
db_customs = LineCustom(
|
|
item_line_id=db_line.id,
|
|
**customs_data.model_dump()
|
|
)
|
|
self.db.add(db_customs)
|
|
|
|
# Create description data if provided
|
|
if description_data:
|
|
db_description = LineDescription(
|
|
item_line_id=db_line.id,
|
|
**description_data.model_dump()
|
|
)
|
|
self.db.add(db_description)
|
|
|
|
# Create reference data if provided
|
|
if reference_data:
|
|
db_reference = LineReference(
|
|
item_line_id=db_line.id,
|
|
**reference_data.model_dump()
|
|
)
|
|
self.db.add(db_reference)
|
|
|
|
return db_line
|
|
|
|
def get_item(self, item_id: int) -> Optional[Item]:
|
|
"""
|
|
Get an item by ID with all nested data loaded
|
|
"""
|
|
item = (
|
|
self.db.query(Item)
|
|
.options(joinedload(Item.lines))
|
|
.filter(Item.id == item_id)
|
|
.first()
|
|
)
|
|
|
|
if not item:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Item with id {item_id} not found"
|
|
)
|
|
|
|
return item
|
|
|
|
def list_items(
|
|
self,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
invoice_id: Optional[int] = None,
|
|
item_type: Optional[str] = None,
|
|
system_origin: Optional[str] = None,
|
|
) -> tuple[list[Item], int]:
|
|
"""
|
|
List items with optional filters and pagination
|
|
Returns tuple of (items, total_count)
|
|
"""
|
|
query = self.db.query(Item).options(joinedload(Item.lines))
|
|
|
|
# Apply filters
|
|
if invoice_id:
|
|
query = query.filter(Item.invoice_id == invoice_id)
|
|
if item_type:
|
|
query = query.filter(Item.item_type == item_type)
|
|
if system_origin:
|
|
query = query.filter(Item.system_origin == system_origin)
|
|
|
|
# Get total count
|
|
total = query.count()
|
|
|
|
# Apply pagination
|
|
items = query.offset(skip).limit(limit).all()
|
|
|
|
return items, total
|
|
|
|
def update_item(self, item_id: int, item_data: ItemUpdate) -> Item:
|
|
"""
|
|
Update an item and optionally its line item (one-to-one)
|
|
"""
|
|
try:
|
|
db_item = self.get_item(item_id)
|
|
|
|
# Extract line data
|
|
line_data = item_data.line
|
|
update_dict = item_data.model_dump(
|
|
exclude={'line'}, exclude_unset=True)
|
|
|
|
# Update item fields
|
|
for field, value in update_dict.items():
|
|
setattr(db_item, field, value)
|
|
|
|
# Update line if provided
|
|
if line_data is not None:
|
|
# Get the existing line or create new one
|
|
db_line = (
|
|
self.db.query(LineItem)
|
|
.filter(LineItem.item_id == item_id)
|
|
.first()
|
|
)
|
|
|
|
if db_line:
|
|
self._update_line_item(db_line, line_data)
|
|
else:
|
|
# Create new line if it doesn't exist
|
|
self._create_line_item(item_id, line_data)
|
|
|
|
self.db.commit()
|
|
self.db.refresh(db_item)
|
|
return db_item
|
|
|
|
except HTTPException:
|
|
raise
|
|
except IntegrityError as e:
|
|
self.db.rollback()
|
|
logger.error(f"Integrity error updating item: {e}")
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Item update failed due to data integrity constraint"
|
|
)
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Unexpected error updating item: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Error updating item: {str(e)}"
|
|
)
|
|
|
|
def _update_line_item(self, db_line: LineItem, line_data: LineItemUpdate):
|
|
"""
|
|
Update a line item and all its nested data
|
|
"""
|
|
# Extract nested data
|
|
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
|
|
)
|
|
|
|
# Update line fields
|
|
for field, value in line_dict.items():
|
|
setattr(db_line, field, value)
|
|
|
|
# Update financial data
|
|
if financial_data:
|
|
db_financial = (
|
|
self.db.query(LineFinancial)
|
|
.filter(LineFinancial.item_line_id == db_line.id)
|
|
.first()
|
|
)
|
|
|
|
if db_financial:
|
|
# Update existing
|
|
for field, value in financial_data.model_dump(exclude_unset=True).items():
|
|
setattr(db_financial, field, value)
|
|
else:
|
|
# Create new
|
|
db_financial = LineFinancial(
|
|
item_line_id=db_line.id,
|
|
**financial_data.model_dump(exclude_unset=True)
|
|
)
|
|
self.db.add(db_financial)
|
|
|
|
# Update quantity data
|
|
if quantity_data:
|
|
db_quantity = (
|
|
self.db.query(LineQuantity)
|
|
.filter(LineQuantity.item_line_id == db_line.id)
|
|
.first()
|
|
)
|
|
|
|
if db_quantity:
|
|
# Update existing
|
|
for field, value in quantity_data.model_dump(exclude_unset=True).items():
|
|
setattr(db_quantity, field, value)
|
|
else:
|
|
# Create new
|
|
db_quantity = LineQuantity(
|
|
item_line_id=db_line.id,
|
|
**quantity_data.model_dump(exclude_unset=True)
|
|
)
|
|
self.db.add(db_quantity)
|
|
|
|
# Update customs data
|
|
if customs_data:
|
|
db_customs = (
|
|
self.db.query(LineCustom)
|
|
.filter(LineCustom.item_line_id == db_line.id)
|
|
.first()
|
|
)
|
|
|
|
if db_customs:
|
|
# Update existing
|
|
for field, value in customs_data.model_dump(exclude_unset=True).items():
|
|
setattr(db_customs, field, value)
|
|
else:
|
|
# Create new
|
|
db_customs = LineCustom(
|
|
item_line_id=db_line.id,
|
|
**customs_data.model_dump(exclude_unset=True)
|
|
)
|
|
self.db.add(db_customs)
|
|
|
|
# Update description data
|
|
if description_data:
|
|
db_description = (
|
|
self.db.query(LineDescription)
|
|
.filter(LineDescription.item_line_id == db_line.id)
|
|
.first()
|
|
)
|
|
|
|
if db_description:
|
|
# Update existing
|
|
for field, value in description_data.model_dump(exclude_unset=True).items():
|
|
setattr(db_description, field, value)
|
|
else:
|
|
# Create new
|
|
db_description = LineDescription(
|
|
item_line_id=db_line.id,
|
|
**description_data.model_dump(exclude_unset=True)
|
|
)
|
|
self.db.add(db_description)
|
|
|
|
# Update reference data
|
|
if reference_data:
|
|
db_reference = (
|
|
self.db.query(LineReference)
|
|
.filter(LineReference.item_line_id == db_line.id)
|
|
.first()
|
|
)
|
|
|
|
if db_reference:
|
|
# Update existing
|
|
for field, value in reference_data.model_dump(exclude_unset=True).items():
|
|
setattr(db_reference, field, value)
|
|
else:
|
|
# Create new
|
|
db_reference = LineReference(
|
|
item_line_id=db_line.id,
|
|
**reference_data.model_dump(exclude_unset=True)
|
|
)
|
|
self.db.add(db_reference)
|
|
|
|
def delete_item(self, item_id: int) -> bool:
|
|
"""
|
|
Delete an item and all its related data (cascade)
|
|
"""
|
|
try:
|
|
db_item = self.get_item(item_id)
|
|
self.db.delete(db_item)
|
|
self.db.commit()
|
|
return True
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Error deleting item: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Error deleting item: {str(e)}"
|
|
)
|
|
|
|
def search_items(
|
|
self,
|
|
search_term: Optional[str] = None,
|
|
skip: int = 0,
|
|
limit: int = 100
|
|
) -> tuple[list[Item], int]:
|
|
"""
|
|
Search items by various fields
|
|
"""
|
|
query = self.db.query(Item).options(joinedload(Item.lines))
|
|
|
|
if search_term:
|
|
search_filter = or_(
|
|
Item.invoice_number.ilike(f"%{search_term}%"),
|
|
Item.reference_number.ilike(f"%{search_term}%"),
|
|
Item.order.ilike(f"%{search_term}%"),
|
|
Item.guide_number.ilike(f"%{search_term}%"),
|
|
)
|
|
query = query.filter(search_filter)
|
|
|
|
total = query.count()
|
|
items = query.offset(skip).limit(limit).all()
|
|
|
|
return items, total
|
|
|
|
# ========================================================================
|
|
# LINE ITEM SPECIFIC OPERATIONS (one-to-one)
|
|
# ========================================================================
|
|
|
|
def get_line_for_item(self, item_id: int) -> Optional[LineItem]:
|
|
"""
|
|
Get the line item for a specific item
|
|
"""
|
|
db_line = (
|
|
self.db.query(LineItem)
|
|
.filter(LineItem.item_id == item_id)
|
|
.first()
|
|
)
|
|
|
|
return db_line
|
|
|
|
def create_or_replace_line(self, item_id: int, line_data: LineItemCreate) -> LineItem:
|
|
"""
|
|
Create or replace the line for an item (one-to-one relationship)
|
|
"""
|
|
try:
|
|
# Verify item exists
|
|
db_item = self.get_item(item_id)
|
|
|
|
# Check if line already exists
|
|
existing_line = (
|
|
self.db.query(LineItem)
|
|
.filter(LineItem.item_id == item_id)
|
|
.first()
|
|
)
|
|
|
|
if existing_line:
|
|
# Delete existing line (cascade will delete financials and quantities)
|
|
self.db.delete(existing_line)
|
|
self.db.flush()
|
|
|
|
# Create new line
|
|
db_line = self._create_line_item(item_id, line_data)
|
|
|
|
self.db.commit()
|
|
self.db.refresh(db_line)
|
|
return db_line
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Error creating/replacing line for item: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Error creating/replacing line for item: {str(e)}"
|
|
)
|
|
|
|
def delete_line_from_item(self, item_id: int) -> bool:
|
|
"""
|
|
Delete the line from an item
|
|
"""
|
|
try:
|
|
db_line = (
|
|
self.db.query(LineItem)
|
|
.filter(LineItem.item_id == item_id)
|
|
.first()
|
|
)
|
|
|
|
if not db_line:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"No line found for item {item_id}"
|
|
)
|
|
|
|
self.db.delete(db_line)
|
|
self.db.commit()
|
|
return True
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Error deleting line from item: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Error deleting line from item: {str(e)}"
|
|
)
|