From c725a6b11a350f2991e49ca33028981739f114bf Mon Sep 17 00:00:00 2001 From: AlexeerCT Date: Thu, 8 Jan 2026 16:25:28 -0600 Subject: [PATCH] feat: Refactor line item schemas and service methods for improved readability and consistency --- .../modules/a76/items/line_items/schemas.py | 196 ++++++++++++------ backend/api/v1/modules/a76/items/service.py | 48 +++-- 2 files changed, 160 insertions(+), 84 deletions(-) diff --git a/backend/api/v1/modules/a76/items/line_items/schemas.py b/backend/api/v1/modules/a76/items/line_items/schemas.py index 771196fb..4921233f 100644 --- a/backend/api/v1/modules/a76/items/line_items/schemas.py +++ b/backend/api/v1/modules/a76/items/line_items/schemas.py @@ -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 diff --git a/backend/api/v1/modules/a76/items/service.py b/backend/api/v1/modules/a76/items/service.py index 4cc4c946..74ba2da1 100644 --- a/backend/api/v1/modules/a76/items/service.py +++ b/backend/api/v1/modules/a76/items/service.py @@ -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))