feat: Refactor line item schemas and service methods for improved readability and consistency

This commit is contained in:
AlexeerCT
2026-01-08 16:25:28 -06:00
parent c040679fc5
commit c725a6b11a
2 changed files with 160 additions and 84 deletions

View File

@@ -6,162 +6,236 @@ from pydantic import BaseModel, Field, ConfigDict, field_validator
from ..line_customs.schemas import (
LineCustomCreate,
LineCustomUpdate,
LineCustomResponse
LineCustomResponse,
)
from ..line_descriptions.schemas import (
LineDescriptionCreate,
LineDescriptionUpdate,
LineDescriptionResponse
LineDescriptionResponse,
)
from ..line_quantities.schemas import (
LineQuantityCreate,
LineQuantityUpdate,
LineQuantityResponse
LineQuantityResponse,
)
from ..line_financials.schemas import (
LineFinancialCreate,
LineFinancialUpdate,
LineFinancialResponse
LineFinancialResponse,
)
from ..line_references.schemas import (
LineReferenceCreate,
LineReferenceUpdate,
LineReferenceResponse
LineReferenceResponse,
)
# ============================================================================
# LINE ITEM SCHEMAS
# ============================================================================
class LineItemBase(BaseModel):
"""Base schema for line items"""
line_number: int = Field(..., description="Line number")
# Part identification
part_number: Optional[str] = Field(None, max_length=50, description="Part number")
component_part_number: Optional[str] = Field(None, max_length=50, description="Component part number")
component_part_number: Optional[str] = Field(
None, max_length=50, description="Component part number"
)
class_code: Optional[str] = Field(None, max_length=20, description="Class code")
@field_validator('class_code', 'part_number', 'component_part_number', 'unit_of_measure', 'alternate_unit', mode='before')
@field_validator(
"class_code",
"part_number",
"component_part_number",
"unit_of_measure",
"alternate_unit",
mode="before",
)
@classmethod
def convert_to_string(cls, v):
"""Convert integers to strings for FK fields"""
if v is not None and not isinstance(v, str):
return str(v)
return v
# Unit of measure
unit_of_measure: Optional[str] = Field(None, max_length=10, description="Unit of measure")
alternate_unit: Optional[str] = Field(None, max_length=10, description="Alternate unit")
unit_of_measure: Optional[str] = Field(
None, max_length=10, description="Unit of measure"
)
alternate_unit: Optional[str] = Field(
None, max_length=10, description="Alternate unit"
)
uma_key: Optional[str] = Field(None, max_length=2, description="UMA key")
auxiliary_unit: Optional[str] = Field(None, max_length=5, description="Auxiliary unit")
auxiliary_unit: Optional[str] = Field(
None, max_length=5, description="Auxiliary unit"
)
# Permits and certificates
permit_number: Optional[str] = Field(None, max_length=20, description="Permit number")
permit_number: Optional[str] = Field(
None, max_length=20, description="Permit number"
)
page_line: Optional[str] = Field(None, max_length=10, description="Page line")
has_certificate: Optional[bool] = Field(None, description="Has certificate")
certificate_number: Optional[str] = Field(None, max_length=10, description="Certificate number")
octave_permit: Optional[str] = Field(None, max_length=20, description="Octave permit")
certificate_number: Optional[str] = Field(
None, max_length=10, description="Certificate number"
)
octave_permit: Optional[str] = Field(
None, max_length=20, description="Octave permit"
)
permits_ped: Optional[str] = Field(None, max_length=500, description="PED permits")
# FDA
has_fda_code: Optional[bool] = Field(None, description="Has FDA code")
fda_key: Optional[str] = Field(None, max_length=10, description="FDA key")
# Subitem flags
is_subitem: Optional[bool] = Field(None, description="Is subitem")
contains_subitems: Optional[bool] = Field(None, description="Contains subitems")
includes_subitems: Optional[bool] = Field(None, description="Includes subitems")
subitem_number: Optional[bool] = Field(None, description="Subitem number")
# Special flags
is_military_mcia: Optional[bool] = Field(None, description="Is military merchandise")
is_military_mcia: Optional[bool] = Field(
None, description="Is military merchandise"
)
# IV32
iv32_type_key: Optional[str] = Field(None, max_length=5, description="IV32 type key")
iv32_type_key: Optional[str] = Field(
None, max_length=5, description="IV32 type key"
)
iv32_number: Optional[str] = Field(None, max_length=35, description="IV32 number")
# Export specific
scrap_invoice: Optional[str] = Field(None, max_length=15, description="Scrap invoice")
consecutive_destination: Optional[int] = Field(None, description="Consecutive destination")
scrap_invoice: Optional[str] = Field(
None, max_length=15, description="Scrap invoice"
)
consecutive_destination: Optional[int] = Field(
None, description="Consecutive destination"
)
ctm_section: Optional[str] = Field(None, max_length=3, description="CTM section")
# Tax payment
tax_payment: Optional[bool] = Field(None, description="Tax payment")
payment_method: Optional[str] = Field(None, max_length=9, description="Payment method")
payment_method: Optional[str] = Field(
None, max_length=9, description="Payment method"
)
igi_amount: Optional[Decimal] = Field(None, description="IGI amount")
igi_payment_method: Optional[str] = Field(None, max_length=9, description="IGI payment method")
igi_payment_method: Optional[str] = Field(
None, max_length=9, description="IGI payment method"
)
# FCC
fcc_key: Optional[str] = Field(None, max_length=30, description="FCC key")
# Valuation method
valuation_method: Optional[str] = Field(None, max_length=2, description="Valuation method")
valuation_determined_value: Optional[Decimal] = Field(None, description="Valuation determined value")
valuation_reason: Optional[str] = Field(None, max_length=500, description="Valuation reason")
valuation_method: Optional[str] = Field(
None, max_length=2, description="Valuation method"
)
valuation_determined_value: Optional[Decimal] = Field(
None, description="Valuation determined value"
)
valuation_reason: Optional[str] = Field(
None, max_length=500, description="Valuation reason"
)
# Container rules
container_rule: Optional[str] = Field(None, max_length=50, description="Container rule")
container_parts_ii: Optional[str] = Field(None, max_length=50, description="Container parts II")
container_rule: Optional[str] = Field(
None, max_length=50, description="Container rule"
)
container_parts_ii: Optional[str] = Field(
None, max_length=50, description="Container parts II"
)
# APHIS
consecutive_aphis: Optional[int] = Field(None, description="Consecutive APHIS")
# BOM/Commercial
bom_version: Optional[int] = Field(None, description="BOM version")
bill_version: Optional[int] = Field(None, description="Bill version")
# TLCAN value
tlcan_value: Optional[Decimal] = Field(None, description="TLCAN value")
# Identifier
identifier: Optional[str] = Field(None, max_length=2, description="Identifier")
# Validation fields
validation_zero: Optional[int] = Field(None, description="Validation zero")
validation_one: Optional[int] = Field(None, description="Validation one")
# Material type
material_type: Optional[str] = Field(None, max_length=50, description="Material type")
material_type: Optional[str] = Field(
None, max_length=50, description="Material type"
)
# Order concept
order_type: Optional[str] = Field(None, max_length=50, description="Order type")
line_concept: Optional[str] = Field(None, max_length=50, description="Line concept")
# Review dispatch
review_dispatch: Optional[str] = Field(None, max_length=10, description="Review dispatch")
review_dispatch: Optional[str] = Field(
None, max_length=10, description="Review dispatch"
)
# Take component from PT
take_component_pt: Optional[int] = Field(None, description="Take component from PT")
# Pallet
pallet2: Optional[int] = Field(None, description="Pallet 2")
# Wildcard field
wildcard_field: Optional[str] = Field(None, max_length=100, description="Wildcard field")
wildcard_field: Optional[str] = Field(
None, max_length=100, description="Wildcard field"
)
class LineItemCreate(LineItemBase):
"""Schema for creating line item with all nested data"""
financial: Optional[LineFinancialCreate] = Field(None, description="Financial data for this line")
quantity: Optional[LineQuantityCreate] = Field(None, description="Quantity data for this line")
customs: Optional[LineCustomCreate] = Field(None, description="Customs data for this line")
description: Optional[LineDescriptionCreate] = Field(None, description="Description data for this line")
reference: Optional[LineReferenceCreate] = Field(None, description="Reference data for this line")
financial: Optional[LineFinancialCreate] = Field(
None, description="Financial data for this line"
)
quantity: Optional[LineQuantityCreate] = Field(
None, description="Quantity data for this line"
)
customs: Optional[LineCustomCreate] = Field(
None, description="Customs data for this line"
)
description: Optional[LineDescriptionCreate] = Field(
None, description="Description data for this line"
)
reference: Optional[LineReferenceCreate] = Field(
None, description="Reference data for this line"
)
class LineItemUpdate(LineItemBase):
"""Schema for updating line item with all nested data"""
line_number: Optional[int] = Field(None, description="Line number")
financial: Optional[LineFinancialUpdate] = Field(None, description="Financial data for this line")
quantity: Optional[LineQuantityUpdate] = Field(None, description="Quantity data for this line")
customs: Optional[LineCustomUpdate] = Field(None, description="Customs data for this line")
description: Optional[LineDescriptionUpdate] = Field(None, description="Description data for this line")
reference: Optional[LineReferenceUpdate] = Field(None, description="Reference data for this line")
financial: Optional[LineFinancialUpdate] = Field(
None, description="Financial data for this line"
)
quantity: Optional[LineQuantityUpdate] = Field(
None, description="Quantity data for this line"
)
customs: Optional[LineCustomUpdate] = Field(
None, description="Customs data for this line"
)
description: Optional[LineDescriptionUpdate] = Field(
None, description="Description data for this line"
)
reference: Optional[LineReferenceUpdate] = Field(
None, description="Reference data for this line"
)
class LineItemResponse(LineItemBase):
"""Schema for line item response with all nested data"""
id: int
item_id: int
financial: Optional[LineFinancialResponse] = None

View File

@@ -36,10 +36,7 @@ class ItemService:
@staticmethod
def get_by_id(
db: Session,
item_id: int,
tenant_id: int,
company_id: int
db: Session, item_id: int, tenant_id: int, company_id: int
) -> Optional[Item]:
"""Get an item by ID with tenant/company validation"""
return (
@@ -91,8 +88,7 @@ class ItemService:
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"])
query = query.filter(Item.system_origin == filters["system_origin"])
if filters.get("search"):
search_term = f"%{filters['search']}%"
query = query.filter(
@@ -186,8 +182,13 @@ class ItemService:
print(f" Has reference: {reference_data is not None}")
line_dict = line_data.model_dump(
exclude={"financial", "quantity",
"customs", "description", "reference"}
exclude={
"financial",
"quantity",
"customs",
"description",
"reference",
}
)
line_dict["item_id"] = db_item.id
line_dict["tenant_id"] = tenant_id
@@ -274,8 +275,7 @@ class ItemService:
# Extract lines data
lines_data = item_data.lines
item_dict = item_data.model_dump(
exclude={"lines"}, exclude_unset=True)
item_dict = item_data.model_dump(exclude={"lines"}, exclude_unset=True)
# Update item fields
for key, value in item_dict.items():
@@ -298,46 +298,48 @@ class ItemService:
reference_data = line_data.reference
line_dict = line_data.model_dump(
exclude={"financial", "quantity",
"customs", "description", "reference"},
exclude_unset=True
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 = 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 = 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 = 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)
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 = reference_data.model_dump(exclude_unset=True)
reference_dict["item_line_id"] = db_line.id
db.add(LineReference(**reference_dict))