Refactor US Tariff Fractions service to use async for get_all method and add USTariffFractionMapper for response mapping

This commit is contained in:
2026-02-27 11:07:16 -06:00
parent e3842337c0
commit 35a6f88211
2 changed files with 37 additions and 3 deletions

View File

@@ -54,8 +54,7 @@ async def list_us_tariff_fractions(
if search:
filters["search"] = search
# Updated to sync call
items, total = USTariffFractionService.get_all(
items, total = await USTariffFractionService.get_all(
db, tenant_id, company_id, skip, page_size, filters
)

View File

@@ -1,4 +1,5 @@
from typing import List, Optional, Tuple, Dict, Any
from datetime import datetime
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
@@ -7,15 +8,49 @@ from decimal import Decimal
from .models import USTariffFraction
from .dto import USTariffFractionCreateDTO, USTariffFractionUpdateDTO
from api.v1.modules.sitar.fracciones_usa.service import FraccionesUSAService
logger = logging.getLogger(__name__)
class USTariffFractionMapper:
"""Maps Sitar FraccionesUSAResponse objects to USTariffFraction domain objects"""
@staticmethod
def to_domain(
item: Any, tenant_id: int, company_id: int
) -> USTariffFraction:
"""Convert a FraccionesUSAResponse to a USTariffFraction instance (not persisted)"""
code = item.FRACCION_CON_PUNTO or item.FRACCION_MOSTRAR or item.FRACCION_SIN_PUNTO or ""
ad_valorem: Optional[float] = None
if item.TARIFA1:
try:
ad_valorem = float(str(item.TARIFA1).replace("%", "").strip())
except (ValueError, TypeError):
ad_valorem = None
fraction = USTariffFraction()
fraction.id = item.CONSECUTIVO
fraction.tenant_id = tenant_id
fraction.company_id = company_id
fraction.code = code
fraction.prefix = item.FRACCION_SIN_PUNTO
fraction.type_code = str(item.NIVEL) if item.NIVEL is not None else None
fraction.ad_valorem = ad_valorem
fraction.fixed_cost = None
fraction.unit_of_measure = item.UNIDADCANTIDAD
fraction.description = item.DESCRIPCION
now = datetime.now()
fraction.created_at = now
fraction.updated_at = now
return fraction
class USTariffFractionService:
"""Service para gestionar fracciones arancelarias americanas"""
@staticmethod
def get_all(
async def get_all(
db: Session,
tenant_id: int,
company_id: int,