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

recontruyendo CRUD en lugar de sitar

Browse Source
This commit is contained in:
Kevin_Ramirez
2026-02-26 12:04:31 -06:00
parent ccb643426d
commit dbb828c92a
2 changed files with 26 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/routes.py
View File

@@ -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)

130
backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/service.py
View File

@@ -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