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

Merge pull request 'feature/columnas_ordenamiento' (#259) from feature/columnas_ordenamiento into development

Browse Source 
Reviewed-on: ADUANASOFT/anexo76#259
This commit is contained in:
acazares
2026-03-30 14:50:10 +00:00
parent 48ee82761e dd4eec56f8
commit 61633fd969
42 changed files with 1838 additions and 1341 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
backend/api/v1/modules/a76/classes/routes.py
View File

@@ -2,7 +2,7 @@
Endpoints API para gestión de clases SCAII y SCAF
"""
from typing import Dict, Any, List
from typing import Dict, Any, List, Optional
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
@@ -33,6 +33,8 @@ async def get_classes_with_fa_data(
company_id: int = Query(..., description="Company ID"),
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(1000, ge=1, le=1000, description="Page size"),
sort_by: Optional[str] = Query(None, description="Column to sort by"),
sort_order: Optional[str] = Query("asc", description="Sort order (asc/desc)"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
@@ -50,6 +52,8 @@ async def get_classes_with_fa_data(
company_id=company_id,
skip=skip,
limit=page_size,
sort_by=sort_by,
sort_order=sort_order,
)
return classes_with_fa

33
backend/api/v1/modules/a76/classes/service.py
View File

@@ -36,6 +36,8 @@ class ClassService:
skip: int = 0,
limit: int = 100,
filters: Optional[Dict[str, Any]] = None,
sort_by: Optional[str] = None,
sort_order: Optional[str] = "asc",
) -> tuple[List[Class], int]:
"""
Get all classes for a tenant with pagination and filters
@@ -68,6 +70,18 @@ class ClassService:
query = query.filter(
Class.physical_review == filters["physical_review"]
)
# Apply sorting
if sort_by:
column = getattr(Class, sort_by, None)
if column:
if sort_order == "desc":
query = query.order_by(column.desc())
else:
query = query.order_by(column.asc())
else:
# Default sorting
query = query.order_by(Class.class_code.asc())
total = query.count()
items = query.offset(skip).limit(limit).all()
@@ -82,6 +96,8 @@ class ClassService:
skip: int = 0,
limit: int = 1000,
filters: Optional[Dict[str, Any]] = None,
sort_by: Optional[str] = None,
sort_order: Optional[str] = "asc",
) -> tuple[List[Dict[str, Any]], int]:
"""
Get all classes with their FA data in a single query using LEFT JOIN.
@@ -125,6 +141,23 @@ class ClassService:
if filters.get("fraction"):
query = query.filter(Class.fraction.ilike(f"%{filters['fraction']}%"))
# Apply sorting
if sort_by:
# Check if sort_by belongs to Class or QClasses
column = getattr(Class, sort_by, None)
if column:
if sort_order == "desc":
query = query.order_by(column.desc())
else:
query = query.order_by(column.asc())
else:
# Handle FA extension fields if sort_by is one of them
# (Simple approach for now, assuming base class fields are prioritized)
pass
else:
# Default sorting
query = query.order_by(Class.class_code.asc())
# Count total before pagination
total = query.count()

13
backend/api/v1/modules/a76/clients_and_providers/service.py
View File

@@ -38,6 +38,8 @@ class ClientProviderService:
skip: int = 0,
limit: int = 50,
filters: Optional[Dict[str, Any]] = None,
sort_by: Optional[str] = None,
sort_order: Optional[str] = "asc",
) -> Tuple[List[ClientProvider], int]:
"""Get all clients/providers for a tenant with pagination"""
query = db.query(ClientProvider).filter(ClientProvider.tenant_id == tenant_id)
@@ -68,6 +70,17 @@ class ClientProviderService:
enabled = 1 if filters["status"] == "enabled" else 0
query = query.filter(ClientProvider.is_active == enabled)
# Apply sorting
if sort_by:
column = getattr(ClientProvider, sort_by, None)
if column:
if sort_order == "desc":
query = query.order_by(column.desc())
else:
query = query.order_by(column.asc())
else:
query = query.order_by(ClientProvider.id.desc())
total = query.count()
clients = (
query.options(

61
backend/api/v1/modules/a76/common/sorting.py Normal file
View File

@@ -0,0 +1,61 @@
from typing import Any, Optional, Type
from sqlalchemy import asc, desc, inspect
from sqlalchemy.orm import Query, RelationshipProperty
def apply_sorting(
query: Query,
model: Type[Any],
sort_by: Optional[str] = None,
sort_desc: bool = True,
default_sort_col: str = "created_at"
) -> Query:
"""
Applies dynamic sorting to a SQLAlchemy query.
Supports nested attributes via dot notation (e.g., 'compliance_mx.remesa').
Automatically handles joins if necessary.
"""
if not sort_by:
if hasattr(model, default_sort_col):
col = getattr(model, default_sort_col)
return query.order_by(desc(col))
return query
try:
parts = sort_by.split('.')
current_model = model
# Traverse relationships if dot notation is used
for i, part in enumerate(parts[:-1]):
# Check if relationship exists
mapper = inspect(current_model)
if part in mapper.relationships:
rel = mapper.relationships[part]
# Join the relationship
query = query.join(rel.entity.class_)
current_model = rel.entity.class_
else:
# If part is not a relationship, we can't go deeper
# Fallback to default sorting
if hasattr(model, default_sort_col):
return query.order_by(desc(getattr(model, default_sort_col)))
return query
# The last part is the actual column
last_part = parts[-1]
if hasattr(current_model, last_part):
col = getattr(current_model, last_part)
if sort_desc:
query = query.order_by(desc(col))
else:
query = query.order_by(asc(col))
else:
# Fallback to default sorting on the original model
if hasattr(model, default_sort_col):
query = query.order_by(desc(getattr(model, default_sort_col)))
except Exception as e:
# Log error or handle gracefully
print(f"Error applying sort for {sort_by}: {e}")
if hasattr(model, default_sort_col):
query = query.order_by(desc(getattr(model, default_sort_col)))
return query

19
backend/api/v1/modules/a76/customs_brokers/services.py
View File

@@ -26,6 +26,8 @@ class CustomsBrokerService:
skip: int = 0,
limit: int = 100,
filters: dict = None,
sort_by: str = None,
sort_order: str = "asc",
):
"""Get all customs brokers for a tenant/company with pagination"""
query = db.query(models.CustomsBroker).filter(
@@ -33,6 +35,23 @@ class CustomsBrokerService:
models.CustomsBroker.company_id == company_id,
)
if filters:
# Implement filters if needed in the future
pass
# Apply sorting
if sort_by:
column = getattr(models.CustomsBroker, sort_by, None)
if column:
if sort_order == "desc":
query = query.order_by(column.desc())
else:
query = query.order_by(column.asc())
else:
query = query.order_by(models.CustomsBroker.id.desc())
else:
query = query.order_by(models.CustomsBroker.id.desc())
total = query.count()
items = query.offset(skip).limit(limit).all()
return items, total

6
backend/api/v1/modules/a76/invoices/routes.py
View File

@@ -1,4 +1,4 @@
from typing import Dict, Any
from typing import Dict, Any, Optional
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
@@ -95,6 +95,8 @@ def list_invoices(
invoice_type: str = Query(None, description="Filter by invoice type"),
manifest_number: str = Query(None, description="Filter by manifest number"),
pedimento: str = Query(None, description="Filter by pedimento"),
sort_by: Optional[str] = Query(None, description="Column to sort by"),
sort_order: Optional[str] = Query("asc", regex="^(asc|desc)$", description="Sort order (asc or desc)"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
@@ -119,7 +121,7 @@ def list_invoices(
filters = {k: v for k, v in filters.items() if v is not None}
items, total = services.InvoiceService.get_all(
db, tenant_id, company_id, skip=skip, limit=page_size, filters=filters
db, tenant_id, company_id, skip=skip, limit=page_size, filters=filters, sort_by=sort_by, sort_order=sort_order
)
print(f"DEBUG: InvoiceService returned {len(items)} items, total={total}")

16
backend/api/v1/modules/a76/invoices/services.py
View File

@@ -115,6 +115,8 @@ class InvoiceService:
skip: int = 0,
limit: int = 100,
filters: Optional[dict] = None,
sort_by: Optional[str] = None,
sort_order: Optional[str] = "asc",
) -> Tuple[List[models.InvoiceHeader], int]:
"""Get all invoices for a tenant/company with pagination and optional filters"""
query = db.query(models.InvoiceHeader).filter(
@@ -165,6 +167,20 @@ class InvoiceService:
models.InvoiceComplianceMx.manifest_number.ilike(f"%{filters['manifest_number']}%")
)
# Apply sorting
if sort_by:
# Simple column mapping
# This can be improved to handle joins if needed
column = getattr(models.InvoiceHeader, sort_by, None)
if column:
if sort_order == "desc":
query = query.order_by(column.desc())
else:
query = query.order_by(column.asc())
else:
# Default sorting if none provided
query = query.order_by(models.InvoiceHeader.id.desc())
total = query.count()
items = query.offset(skip).limit(limit).all()

28
backend/api/v1/modules/a76/items/exports/validators/create.py
View File

@@ -53,27 +53,28 @@ def validate_create(
if not line.class_id:
errors.add_required_error(field=f"line[{line_number}].class_id")
if not line.quantity.quantity or line.quantity.quantity <= 0:
if not line.quantity or not line.quantity.quantity or line.quantity.quantity <= 0:
errors.add_required_error(field=f"line[{line_number}].quantity.quantity")
# TODO: Añadir validacion SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf <-- de la tabla de preferencias de el sistema
# if SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf == False:
if fa_data and not fa_data.is_subitem:
if (
not line.financial.unit_cost_capture
not line.financial
or not line.financial.unit_cost_capture
or line.financial.unit_cost_capture <= 0
):
errors.add_required_error(
field=f"line[{line_number}].financial.unit_cost_capture"
)
if not line.quantity.net_weight or line.quantity.net_weight <= 0:
if not line.quantity or not line.quantity.net_weight or line.quantity.net_weight <= 0:
errors.add_required_error(field=f"line[{line_number}].quantity.net_weight")
if not line.customs.origin_country:
if not line.customs or not line.customs.origin_country:
errors.add_required_error(field=f"line[{line_number}].customs.origin_country")
if not line.customs.fraction_type:
if not line.customs or not line.customs.fraction_type:
errors.add_required_error(field=f"line[{line_number}].customs.fraction_type")
# FA-specific validations
@@ -142,6 +143,23 @@ def validate_create(
validate_common(db, line, tenant_id, company_id, errors, line_number)
if not errors.has_errors():
# Ensure nested objects exist for calculations
from ...schemas import (
LineFinancialCreate,
LineQuantityCreate,
LineCustomCreate,
LineDescriptionCreate,
)
if line.financial is None:
line.financial = LineFinancialCreate()
if line.quantity is None:
line.quantity = LineQuantityCreate()
if line.customs is None:
line.customs = LineCustomCreate()
if line.description is None:
line.description = LineDescriptionCreate()
# Obtener la factura para acceder a tipo de cambio, moneda y peso
invoice: InvoiceHeader = (
db.query(InvoiceHeader)

17
backend/api/v1/modules/a76/items/exports/validators/update.py
View File

@@ -27,6 +27,23 @@ def validate_update(
validate_common(db, line, tenant_id, company_id, errors, line_number)
if not errors.has_errors():
# Ensure nested objects exist for calculations/partial updates
from ...schemas import (
LineFinancialCreate,
LineQuantityCreate,
LineCustomCreate,
LineDescriptionCreate,
)
if line.financial is None:
line.financial = LineFinancialCreate()
if line.quantity is None:
line.quantity = LineQuantityCreate()
if line.customs is None:
line.customs = LineCustomCreate()
if line.description is None:
line.description = LineDescriptionCreate()
# Obtener la factura para acceder a tipo de cambio, moneda y peso
invoice: InvoiceHeader = (
db.query(InvoiceHeader)

33
backend/api/v1/modules/a76/items/imports/validators/common.py
View File

@@ -61,7 +61,7 @@ def validate_common(
code="UNIT_OF_MEASURE_REQUIRED",
)
if not line.customs.fraction:
if not line.customs or not line.customs.fraction:
if not line_item:
if not class_.fraction:
errors.add_error(
@@ -73,7 +73,7 @@ def validate_common(
else:
fraction = class_.fraction
else:
if not line.customs.fraction:
if not line.customs or not line.customs.fraction:
if not class_.fraction:
errors.add_error(
field=f"line[{line_number}].customs.fraction",
@@ -87,7 +87,7 @@ def validate_common(
if line_item:
fraction = line.customs.fraction
if not line.description.description_spanish and not class_.description_es:
if (not line.description or not line.description.description_spanish) and not class_.description_es:
errors.add_error(
field=f"line[{line_number}].description.description_spanish",
message="La descripción en español es obligatoria para la clase especificada.",
@@ -95,7 +95,7 @@ def validate_common(
code="DESCRIPTION_SPANISH_REQUIRED",
)
if not line.description.description_english and not class_.description_en:
if (not line.description or not line.description.description_english) and not class_.description_en:
errors.add_error(
field=f"line[{line_number}].description.description_english",
message="La descripción en inglés es obligatoria para la clase especificada.",
@@ -103,7 +103,7 @@ def validate_common(
code="DESCRIPTION_ENGLISH_REQUIRED",
)
if line.quantity.quantity and line.quantity.quantity <= 0:
if line.quantity and line.quantity.quantity and line.quantity.quantity <= 0:
errors.add_error(
field=f"line[{line_number}].quantity.quantity",
message="La cantidad debe ser mayor a cero.",
@@ -129,7 +129,7 @@ def validate_common(
code="UNIT_OF_MEASURE_NOT_FOUND",
)
if line.quantity.package_id:
if line.quantity and line.quantity.package_id:
package = (
db.query(func.count(Package.id))
.filter(
@@ -161,7 +161,7 @@ def validate_common(
code="PACKAGE_QUANTITY_MUST_BE_GREATER_THAN_ZERO",
)
else:
if line.quantity.package_quantity and (line.quantity.package_quantity > 0 and not line.quantity.package_id):
if line.quantity and line.quantity.package_quantity and (line.quantity.package_quantity > 0 and not line.quantity.package_id):
errors.add_error(
field=f"line[{line_number}].quantity.package_id",
message="El paquete es obligatorio cuando se proporciona la cantidad de paquetes.",
@@ -173,9 +173,9 @@ def validate_common(
fraction_type = None
sector = None
if fraction:
fraction = line.customs.fraction if line.customs.fraction else fraction
fraction = (line.customs.fraction if line.customs.fraction else fraction) if line.customs else fraction
country = line.customs.origin_country
country = line.customs.origin_country if line.customs else None
if line_item and line_item.customs:
country = (
line_item.customs.origin_country
@@ -183,7 +183,7 @@ def validate_common(
else country
)
fraction_type = line.customs.fraction_type.upper()
fraction_type = (line.customs.fraction_type.upper() if line.customs.fraction_type else None) if line.customs else None
if line_item and line_item.customs:
fraction_type = (
line_item.customs.fraction_type
@@ -191,7 +191,7 @@ def validate_common(
else fraction_type
)
sector = line.customs.sector
sector = line.customs.sector if line.customs else None
if line_item and line_item.customs:
sector = line_item.customs.sector if line_item.customs.sector else sector
@@ -210,7 +210,7 @@ def validate_common(
code="ORIGIN_COUNTRY_NOT_FOUND",
)
else:
if fraction_type.strip().upper() not in vars(FractionType).values():
if fraction_type and fraction_type.strip().upper() not in vars(FractionType).values():
valid_types = [
v
for k, v in vars(FractionType).items()
@@ -225,15 +225,16 @@ def validate_common(
code="FRACTION_TYPE_INVALID",
value=fraction_type,
)
else:
if fraction_type.strip().upper() == FractionType.PROSEC and not sector:
elif fraction_type:
ft_upper = fraction_type.strip().upper()
if ft_upper == FractionType.PROSEC and not sector:
errors.add_error(
field=f"line[{line_number}].customs.sector",
message="El sector es obligatorio cuando el tipo de fracción es 'PROSEC'.",
solution=["Proporciona un sector valido."],
code="SECTOR_REQUIRED_FOR_PROSEC",
)
elif fraction_type.strip().upper() != FractionType.PROSEC and sector:
elif ft_upper != FractionType.PROSEC and sector:
errors.add_error(
field=f"line[{line_number}].customs.sector",
message="El sector solo es aplicable cuando el tipo de fracción es 'PROSEC'.",
@@ -242,7 +243,7 @@ def validate_common(
],
code="SECTOR_ONLY_FOR_PROSEC",
)
elif fraction_type.strip().upper() == FractionType.PROSEC and sector:
elif ft_upper == FractionType.PROSEC and sector:
sector_db: Sector = (
db.query(Sector).filter(
Sector.key == sector,

28
backend/api/v1/modules/a76/items/imports/validators/create.py
View File

@@ -52,27 +52,28 @@ def validate_create(
if not line.class_id:
errors.add_required_error(field=f"line[{line_number}].class_id")
if not line.quantity.quantity or line.quantity.quantity <= 0:
if not line.quantity or not line.quantity.quantity or line.quantity.quantity <= 0:
errors.add_required_error(field=f"line[{line_number}].quantity.quantity")
# TODO: Añadir validacion SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf <-- de la tabla de preferencias de el sistema
# if SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf == False:
if fa_data and not fa_data.is_subitem:
if (
not line.financial.unit_cost_capture
not line.financial
or not line.financial.unit_cost_capture
or line.financial.unit_cost_capture <= 0
):
errors.add_required_error(
field=f"line[{line_number}].financial.unit_cost_capture"
)
if not line.quantity.net_weight or line.quantity.net_weight <= 0:
if not line.quantity or not line.quantity.net_weight or line.quantity.net_weight <= 0:
errors.add_required_error(field=f"line[{line_number}].quantity.net_weight")
if not line.customs.origin_country:
if not line.customs or not line.customs.origin_country:
errors.add_required_error(field=f"line[{line_number}].customs.origin_country")
if not line.customs.fraction_type:
if not line.customs or not line.customs.fraction_type:
errors.add_required_error(field=f"line[{line_number}].customs.fraction_type")
# FA-specific validations
@@ -126,6 +127,23 @@ def validate_create(
validate_common(db, line, tenant_id, company_id, errors, line_number)
if not errors.has_errors():
# Ensure nested objects exist for calculations
from ...schemas import (
LineFinancialCreate,
LineQuantityCreate,
LineCustomCreate,
LineDescriptionCreate,
)
if line.financial is None:
line.financial = LineFinancialCreate()
if line.quantity is None:
line.quantity = LineQuantityCreate()
if line.customs is None:
line.customs = LineCustomCreate()
if line.description is None:
line.description = LineDescriptionCreate()
# Obtener la factura para acceder a tipo de cambio, moneda y peso
invoice: InvoiceHeader = (
db.query(InvoiceHeader)

17
backend/api/v1/modules/a76/items/imports/validators/update.py
View File

@@ -26,6 +26,23 @@ def validate_update(
validate_common(db, line, tenant_id, company_id, errors, line_number)
if not errors.has_errors():
# Ensure nested objects exist for calculations/partial updates
from ...schemas import (
LineFinancialCreate,
LineQuantityCreate,
LineCustomCreate,
LineDescriptionCreate,
)
if line.financial is None:
line.financial = LineFinancialCreate()
if line.quantity is None:
line.quantity = LineQuantityCreate()
if line.customs is None:
line.customs = LineCustomCreate()
if line.description is None:
line.description = LineDescriptionCreate()
# Obtener la factura para acceder a tipo de cambio, moneda y peso
invoice: InvoiceHeader = (
db.query(InvoiceHeader)

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

@@ -83,6 +83,8 @@ async def list_items(
None, description="Filter by system origin (SCAF/SCAII)"),
search: Optional[str] = Query(
None, description="Search term for invoice number, reference, order, or guide"),
sort_by: Optional[str] = Query(None, description="Column to sort by"),
sort_order: Optional[str] = Query("asc", regex="^(asc|desc)$", description="Sort order (asc or desc)"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
@@ -107,7 +109,8 @@ async def list_items(
}
items, total = service.get_all(
db, tenant_id, company_id, skip, limit, filters)
db, tenant_id, company_id, skip, limit, filters, sort_by=sort_by, sort_order=sort_order
)
return LineItemListResponse(
total=total,
@@ -168,23 +171,26 @@ async def delete_item(
# ADDITIONAL ENDPOINTS FOR INVOICE
# ============================================================================
@router.get("/invoice/{invoice_id}/items", response_model=LineItemListResponse)
async def get_items_by_invoice(
@router.get("/invoice/{invoice_id}/items/", response_model=LineItemListResponse)
async def list_items_by_invoice(
invoice_id: int = Path(..., description="Invoice ID"),
company_id: int = Query(..., description="Company ID"),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Maximum records to return"),
sort_by: Optional[str] = Query(None, description="Column to sort by"),
sort_order: Optional[str] = Query("asc", regex="^(asc|desc)$", description="Sort order (asc or desc)"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""
Get all items for a specific invoice
List all items for a specific invoice
"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
service = ItemService()
items, total = service.get_by_invoice(
db, invoice_id, tenant_id, company_id, skip, limit)
db, invoice_id, tenant_id, company_id, skip, limit, sort_by=sort_by, sort_order=sort_order
)
return LineItemListResponse(
total=total,

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

@@ -278,6 +278,8 @@ class ItemService:
skip: int = 0,
limit: int = 100,
filters: Optional[dict] = None,
sort_by: Optional[str] = None,
sort_order: Optional[str] = "asc",
) -> Tuple[List[LineItem], int]:
"""Get all items for a tenant/company with pagination and optional filters"""
query = (
@@ -316,6 +318,25 @@ class ItemService:
LineItem.guide_number.ilike(search_term),
)
)
if filters.get("invoice_number"):
# Join with InvoiceHeader to search by invoice_number
query = query.join(InvoiceHeader).filter(
InvoiceHeader.invoice_number.ilike(f"%{filters['invoice_number']}%")
)
# Apply sorting
if sort_by:
# Map sort_by to actual model column if possible
# Note: Some columns might require joins if they are in related models
column = getattr(LineItem, sort_by, None)
if column:
if sort_order == "desc":
query = query.order_by(column.desc())
else:
query = query.order_by(column.asc())
else:
# Default sorting
query = query.order_by(LineItem.line_number.asc())
total = query.count()
items = query.offset(skip).limit(limit).all()
@@ -332,6 +353,8 @@ class ItemService:
company_id: int,
skip: int = 0,
limit: int = 100,
sort_by: Optional[str] = None,
sort_order: Optional[str] = "asc",
) -> Tuple[List[LineItem], int]:
"""Get all items for a specific invoice"""
query = (
@@ -351,6 +374,18 @@ class ItemService:
)
)
# Apply sorting
if sort_by:
column = getattr(LineItem, sort_by, None)
if column:
if sort_order == "desc":
query = query.order_by(column.desc())
else:
query = query.order_by(column.asc())
else:
# Default sorting
query = query.order_by(LineItem.line_number.asc())
total = query.count()
items = query.offset(skip).limit(limit).all()
for item in items:
@@ -494,8 +529,13 @@ class ItemService:
)
except Exception as e:
db.rollback()
logger.error(f"Unexpected error creating LineItem: {e}")
raise HTTPException(status_code=500, detail="Error creating LineItem")
import traceback
error_msg = f"Unexpected error creating LineItem: {str(e)}"
logger.error(f"{error_msg}\n{traceback.format_exc()}")
raise HTTPException(
status_code=500,
detail=f"Error interno al crear la partida: {type(e).__name__}: {str(e)}"
)
@staticmethod
def update(

14
backend/api/v1/modules/a76/parts/service.py
View File

@@ -55,6 +55,8 @@ class PartService:
skip: int = 0,
limit: int = 100,
filters: Optional[Dict[str, Any]] = None,
sort_by: Optional[str] = None,
sort_order: Optional[str] = "asc",
) -> tuple[List[Part], int]:
query = db.query(Part).filter(Part.tenant_id == tenant_id)
@@ -82,6 +84,18 @@ class PartService:
)
)
# Otros filtros...
# Apply sorting
if sort_by:
column = getattr(Part, sort_by, None)
if column:
if sort_order == "desc":
query = query.order_by(column.desc())
else:
query = query.order_by(column.asc())
else:
# Default sorting: Newest first
query = query.order_by(Part.id.desc())
total = query.count()
items = query.offset(skip).limit(limit).all()

20
backend/api/v1/modules/a76/pedmientos/services/pedimentos.py
View File

@@ -69,6 +69,8 @@ class PedimentosService:
skip: int = 0,
limit: int = 100,
filters: Optional[Dict[str, Any]] = None,
sort_by: Optional[str] = None,
sort_order: Optional[str] = "asc",
) -> tuple[List[Pedimentos], int]:
"""
Get all pedimentos for a tenant with pagination and filters
@@ -126,12 +128,22 @@ class PedimentosService:
selectinload(Pedimentos.pedimento_seals),
selectinload(Pedimentos.pedimento_containers),
)
.order_by(desc(Pedimentos.created_at))
.offset(skip)
.limit(limit)
.all()
)
# Apply sorting
if sort_by:
column = getattr(Pedimentos, sort_by, None)
if column:
if sort_order == "desc":
query = query.order_by(column.desc())
else:
query = query.order_by(column.asc())
else:
# Default sorting
query = query.order_by(desc(Pedimentos.created_at))
items = query.offset(skip).limit(limit).all()
return items, total
@staticmethod