ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor item management: Update schemas, models, and API routes

Browse Source 
- Renamed customs-related schemas in line_items to LineCustomCreate, LineCustomUpdate, and LineCustomResponse.
- Adjusted models to streamline invoice_id mapping in Item model.
- Enhanced item routes to include company_id in summary statistics endpoint.
- Refactored ItemService to improve item creation and update logic, removing redundant methods.
- Updated frontend components for item management, including new item creation and editing functionalities.
- Added API client for items with CRUD operations and improved error handling.
This commit is contained in:
AlexeerCT
2025-12-30 09:22:52 -06:00
parent 304f7b07d4
commit 30b8daf16e
8 changed files with 623 additions and 481 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
backend/api/v1/modules/a76/items/line_items/schemas.py
View File

@@ -4,9 +4,9 @@ from pydantic import BaseModel, Field, ConfigDict
# Import nested schemas
from ..line_customs.schemas import (
LineCustomsCreate,
LineCustomsUpdate,
LineCustomsResponse
LineCustomCreate,
LineCustomUpdate,
LineCustomResponse
)
from ..line_descriptions.schemas import (
LineDescriptionCreate,
@@ -137,7 +137,7 @@ 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[LineCustomsCreate] = Field(None, description="Customs 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")
@@ -147,7 +147,7 @@ class LineItemUpdate(LineItemBase):
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[LineCustomsUpdate] = Field(None, description="Customs 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")
@@ -158,7 +158,7 @@ class LineItemResponse(LineItemBase):
item_id: int
financial: Optional[LineFinancialResponse] = None
quantity: Optional[LineQuantityResponse] = None
customs: Optional[LineCustomsResponse] = None
customs: Optional[LineCustomResponse] = None
description: Optional[LineDescriptionResponse] = None
reference: Optional[LineReferenceResponse] = None

3
backend/api/v1/modules/a76/items/models.py
View File

@@ -28,8 +28,7 @@ class Item(Base, TenantScopedMixin, TimestampMixin):
}
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
invoice_id: Mapped[int] = mapped_column(
ForeignKey("a76.invoice_header.id")) # CONSECUTIVO
invoice_id: Mapped[int] = mapped_column(ForeignKey("a76.invoice_header.id")) # CONSECUTIVO
# Type: IMPORT_TEMP, IMPORT_DEF, EXPORT, REPAIR, etc.
item_type: Mapped[str] = mapped_column(String(20))
system_origin: Mapped[str] = mapped_column(String(10)) # SCAF or SCAII

12
backend/api/v1/modules/a76/items/routes.py
View File

@@ -198,6 +198,7 @@ async def get_items_by_invoice(
@router.get("/stats/summary")
async def get_items_summary(
company_id: int = Query(..., description="Company ID"),
invoice_id: Optional[int] = Query(
None, description="Filter by invoice ID"),
db: Session = Depends(get_core_db),
@@ -206,11 +207,16 @@ async def get_items_summary(
"""
Get summary statistics for items
"""
service = ItemService(db)
items, total = service.list_items(
tenant_id = validate_access_to_resource(db, company_id, current_user)
filters = {"invoice_id": invoice_id} if invoice_id else None
items, total = ItemService.get_all(
db=db,
tenant_id=tenant_id,
company_id=company_id,
skip=0,
limit=10000, # Get all for stats
invoice_id=invoice_id
filters=filters
)
# Calculate stats

454
backend/api/v1/modules/a76/items/service.py
View File

@@ -344,457 +344,3 @@ class ItemService:
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)}"
)

2
backend/api/v1/modules/a76/router.py
View File

@@ -9,6 +9,7 @@ from .customs_brokers.routes import router as customs_broker_router
# Importar routers de módulos
from .invoices.routes import router as invoices_router
from .items.routes import router as items_router
from .classes import router as classes_router
from .clients_and_providers import router as client_and_provider_router
from .general_catalogs.company import router as company_router
@@ -46,6 +47,7 @@ router = APIRouter()
# Registrar módulos
router.include_router(invoices_router, prefix="/a76", tags=["a76 / invoices"])
router.include_router(items_router, prefix="/a76", tags=["a76 / items"])
router.include_router(pedimentos_router, prefix="/a76")
router.include_router(
client_and_provider_router, prefix="/a76", tags=["a76 / clients_and_providers"]