diff --git a/backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/routes.py b/backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/routes.py index d5552c7e..daabcd8a 100644 --- a/backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/routes.py +++ b/backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/routes.py @@ -17,10 +17,19 @@ from .dto import ( ) from .service import USTariffFractionService -# Create base router with generic CRUD routes - REMOVED strictly read-only from Sitar -# Writes are disabled at API level, but Service still supports fallback writes if needed internally +# Create router using TenantCRUDRoutes factory for basic CRUD operations +crud_router = TenantCRUDRoutes( + service=USTariffFractionService, + create_schema=USTariffFractionCreateDTO, + update_schema=USTariffFractionUpdateDTO, + response_schema=USTariffFractionResponseDTO, + prefix="/us-tariff-fractions", + tags=["a76 / general catalogs / us tariff fractions"], + resource_name="US Tariff Fraction", + enable_list=False, # We implement our custom list endpoint +) -router = APIRouter(prefix="/us-tariff-fractions", tags=["a76 / general catalogs / us tariff fractions"]) +router = crud_router.router # Custom list endpoint with search filter @router.get( @@ -44,8 +53,8 @@ async def list_us_tariff_fractions( if search: filters["search"] = search - # Updated to async call with Sitar integration - items, total = await USTariffFractionService.get_all( + # Updated to sync call + items, total = USTariffFractionService.get_all( db, tenant_id, company_id, skip, page_size, filters ) @@ -56,24 +65,3 @@ async def list_us_tariff_fractions( "page_size": page_size, "pages": (total + page_size - 1) // page_size, } - - -@router.get( - "/{us_tariff_fraction_id}", - response_model=USTariffFractionResponseDTO, - summary="Get US Tariff Fraction by ID", - description="Get a specific US tariff fraction by ID (Lookups in Local DB for legacy compatibility)", -) -async def get_us_tariff_fraction( - us_tariff_fraction_id: int, - company_id: int = Query(..., description="Company ID"), - db: Session = Depends(get_core_db), - current_user: Dict[str, Any] = Depends(get_current_user), -): - tenant_id = validate_access_to_resource(db, company_id, current_user) - - item = USTariffFractionService.get_by_id(db, tenant_id, company_id, us_tariff_fraction_id) - if not item: - from fastapi import HTTPException - raise HTTPException(status_code=404, detail="US Tariff fraction not found") - return USTariffFractionResponseDTO.model_validate(item) diff --git a/backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/service.py b/backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/service.py index 877dfcc5..42e8b956 100644 --- a/backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/service.py +++ b/backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/service.py @@ -1,73 +1,21 @@ -""" -Service para fracciones arancelarias americanas -""" - from typing import List, Optional, Tuple, Dict, Any from sqlalchemy.orm import Session from sqlalchemy.exc import IntegrityError from fastapi import HTTPException -import zlib import logging -import re from decimal import Decimal from .models import USTariffFraction from .dto import USTariffFractionCreateDTO, USTariffFractionUpdateDTO -from api.v1.modules.sitar.fracciones_usa.service import FraccionesUSAService -from api.v1.modules.sitar.fracciones_usa.schemas import FraccionesUSAResponse logger = logging.getLogger(__name__) -class USTariffFractionMapper: - """Helper to map Sitar USA responses to Local domain objects""" - - @staticmethod - def to_domain(fraccion: FraccionesUSAResponse, tenant_id: int, company_id: int) -> USTariffFraction: - # Generate a deterministic numeric ID based on the unique code - # We use CRC32 to get a consistent integer implementation-independent - fake_id = zlib.crc32((fraccion.FRACCION_SIN_PUNTO or "").encode('utf-8')) - - # Parse numeric values safely - ad_valorem = None - if fraccion.TARIFA1: - try: - # Extract numbers from string like "5.2%" or similar if present - # Assuming TARIFA1 might be clean number or percentage string - clean_val = re.sub(r'[^\d.]', '', str(fraccion.TARIFA1)) - if clean_val: - ad_valorem = Decimal(clean_val) - except: - pass - - fixed_cost = None - if fraccion.ESPECIFICO: - try: - clean_val = re.sub(r'[^\d.]', '', str(fraccion.ESPECIFICO)) - if clean_val: - fixed_cost = Decimal(clean_val) - except: - pass - - return USTariffFraction( - id=fake_id, # Updated to use fake_id instead of sitar consecutive if needed, or consistent hash - tenant_id=tenant_id, - company_id=company_id, - code=fraccion.FRACCION_SIN_PUNTO or "", - prefix=None, # Not mapped from Sitar response currently - type_code=None, - ad_valorem=ad_valorem, - fixed_cost=fixed_cost, - unit_of_measure=fraccion.UNIDADCANTIDAD, - description=fraccion.DESCRIPCION - ) - - class USTariffFractionService: """Service para gestionar fracciones arancelarias americanas""" @staticmethod - async def get_all( + def get_all( db: Session, tenant_id: int, company_id: int, @@ -75,60 +23,7 @@ class USTariffFractionService: limit: int = 100, filters: Optional[Dict[str, Any]] = None, ) -> Tuple[List[USTariffFraction], int]: - """ - Obtiene todas las fracciones arancelarias americanas con filtros opcionales. - Estrategia: Sitar API -> Fallback Local DB - """ - - # 1. Try Sitar API - try: - sitar_service = FraccionesUSAService.get_instance() - - sitar_fraccion = None - has_filters = False - - if filters and filters.get("search"): - term = filters["search"] - # Sitar only filters by fraction code - if term.replace(".", "").isdigit(): - sitar_fraccion = term - has_filters = True - - sitar_items = await sitar_service.search( - fraccion=sitar_fraccion, - skip=skip, - limit=limit - ) - - # If Sitar returns empty list AND we didn't have specific filters, attempt fallback - if not sitar_items and not has_filters: - logger.warning("Sitar return empty list for USA broad query. Attempting fallback to local DB.") - return USTariffFractionService._get_all_local(db, tenant_id, company_id, skip, limit, filters) - - # Map items - items = [USTariffFractionMapper.to_domain(item, tenant_id, company_id) for item in sitar_items] - - # Estimate total - total = len(items) + skip - if len(items) == limit: - total += 1 - - return items, total - - except Exception as e: - logger.error(f"Error fetching USA Fractions from Sitar API, falling back to local DB: {e}") - return USTariffFractionService._get_all_local(db, tenant_id, company_id, skip, limit, filters) - - @staticmethod - def _get_all_local( - db: Session, - tenant_id: int, - company_id: int, - skip: int = 0, - limit: int = 100, - filters: Optional[Dict[str, Any]] = None, - ) -> Tuple[List[USTariffFraction], int]: - """Lógica original de consulta local""" + """Obtiene todas las fracciones locales con filtros opcionales.""" query = db.query(USTariffFraction).filter( USTariffFraction.tenant_id == tenant_id, USTariffFraction.company_id == company_id, @@ -150,13 +45,14 @@ class USTariffFractionService: return items, total + return items, total + @staticmethod def get_by_id( - db: Session, tenant_id: int, company_id: int, fraction_id: int + db: Session, fraction_id: int, tenant_id: int, company_id: int ) -> Optional[USTariffFraction]: """ - Obtiene por ID. - Legacy: Consulta Local DB. + Obtiene por ID local. """ return ( db.query(USTariffFraction) @@ -168,14 +64,12 @@ class USTariffFractionService: .first() ) - # WRITE OPERATIONS - DEPRECATED / LOCAL ONLY - @staticmethod def create( db: Session, + fraction_data: USTariffFractionCreateDTO, tenant_id: int, company_id: int, - fraction_data: USTariffFractionCreateDTO, ) -> USTariffFraction: try: db_fraction = USTariffFraction( @@ -198,13 +92,13 @@ class USTariffFractionService: @staticmethod def update( db: Session, - tenant_id: int, - company_id: int, fraction_id: int, + tenant_id: int, fraction_data: USTariffFractionUpdateDTO, + company_id: int, ) -> Optional[USTariffFraction]: db_fraction = USTariffFractionService.get_by_id( - db, tenant_id, company_id, fraction_id + db, fraction_id, tenant_id, company_id ) if not db_fraction: return None @@ -219,10 +113,10 @@ class USTariffFractionService: @staticmethod def delete( - db: Session, tenant_id: int, company_id: int, fraction_id: int + db: Session, fraction_id: int, tenant_id: int, company_id: int ) -> bool: db_fraction = USTariffFractionService.get_by_id( - db, tenant_id, company_id, fraction_id + db, fraction_id, tenant_id, company_id ) if not db_fraction: return False