Merge remote-tracking branch 'origin/development' into fix/clases_activo_fijo

This commit is contained in:
2026-02-27 10:54:27 -06:00
19 changed files with 188 additions and 117 deletions

View File

@@ -8,7 +8,7 @@ from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from core.database import get_core_db
from core.security import get_current_user
from core.security import get_current_user, get_tenant_from_token
from .dto import (
TariffFractionCreateDTO,
@@ -27,6 +27,7 @@ router = APIRouter(prefix="/tariff-fractions", tags=["a76 / general catalogs / t
description="Get paginated list of Tariff Fractions with optional search filter (global catalog)",
)
async def list_tariff_fractions(
company_id: int = Query(..., description="Company ID"),
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(50, ge=1, le=10000, description="Page size"),
search: Optional[str] = Query(None, description="Search in code, fraction, description, nico, or umt"),
@@ -47,8 +48,9 @@ async def list_tariff_fractions(
# Service.get_all calls Sitar (async) or DB (sync).
# This should be fine.
tenant_id = current_user.get("tenant_id")
company_id = current_user.get("company_id") # Assuming user is context-aware or we use a default?
tenant_id = get_tenant_from_token(current_user)
if tenant_id is None:
tenant_id = current_user.get("tenant_id")
# If using headers for selected company, it might be in current_user context if middleware sets it.
items, total = await TariffFractionService.get_all(
@@ -121,7 +123,7 @@ async def create_tariff_fraction(
pass
us_dto = USTariffFractionCreateDTO(
code=fraction_data.code,
code=fraction_data.fraction, # Store the punctuated fraction in the DB
description=fraction_data.description,
unit_of_measure=fraction_data.umt,
ad_valorem=ad_valorem,
@@ -131,7 +133,7 @@ async def create_tariff_fraction(
fixed_cost=None
)
created = USTariffFractionService.create(db, tenant_id, company_id, us_dto)
created = USTariffFractionService.create(db, us_dto, tenant_id, company_id)
return TariffFractionService.to_domain_usa_local(created)
else:
@@ -171,12 +173,13 @@ async def update_tariff_fraction(
pass
us_dto = USTariffFractionUpdateDTO(
code=fraction_data.fraction,
description=fraction_data.description,
unit_of_measure=fraction_data.umt,
ad_valorem=ad_valorem
)
updated = USTariffFractionService.update(db, tenant_id, company_id, tariff_fraction_id, us_dto)
updated = USTariffFractionService.update(db, tariff_fraction_id, tenant_id, us_dto, company_id)
if not updated:
raise HTTPException(status_code=404, detail="US Tariff fraction not found")
return TariffFractionService.to_domain_usa_local(updated)
@@ -202,7 +205,7 @@ async def delete_tariff_fraction(
if catalog == "american":
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.service import USTariffFractionService
success = USTariffFractionService.delete(db, tenant_id, company_id, tariff_fraction_id)
success = USTariffFractionService.delete(db, tariff_fraction_id, tenant_id, company_id)
if not success:
raise HTTPException(status_code=404, detail="US Tariff fraction not found")
return {"ok": True}

View File

@@ -96,10 +96,14 @@ class TariffFractionService:
# US format: 1234.56.78.90. For now return as is or use helper if available.
# item is USTariffFraction (imported inside method to avoid circular import if needed, or assumed available)
# Remove formatting (e.g. dots) for the 'code' property
code_str = str(item.code)
clean_code = code_str.replace(".", "").replace("-", "")
return TariffFraction(
id=item.id,
code=item.code,
fraction=item.code, # TODO: Format if needed
code=clean_code,
fraction=code_str,
description=item.description or "(Sin descripción)",
nico=None,
umt=item.unit_of_measure,
@@ -133,11 +137,11 @@ class TariffFractionService:
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.service import USTariffFractionService
# Use local service directly
usa_items, total = USTariffFractionService._get_all_local(
usa_items, total = USTariffFractionService.get_all(
db, tenant_id, company_id, skip, limit, filters
)
items = [TariffFractionMapper.to_domain_usa_local(item) for item in usa_items]
items = [TariffFractionService.to_domain_usa_local(item) for item in usa_items]
return items, total
# USA CATALOG HANDLING (API - 'Fracciones US')

View File

@@ -3,9 +3,9 @@ DTOs para fracciones arancelarias americanas
"""
from datetime import datetime
from typing import Optional
from typing import Optional, Any
from pydantic import BaseModel, Field, ConfigDict
from pydantic import BaseModel, Field, ConfigDict, model_validator
class USTariffFractionCreateDTO(BaseModel):
@@ -23,6 +23,7 @@ class USTariffFractionCreateDTO(BaseModel):
class USTariffFractionUpdateDTO(BaseModel):
"""DTO para actualizar fracción arancelaria americana"""
code: Optional[str] = Field(None, max_length=16)
prefix: Optional[str] = Field(None, max_length=10)
type_code: Optional[str] = Field(None, max_length=10)
ad_valorem: Optional[float] = None
@@ -38,6 +39,7 @@ class USTariffFractionResponseDTO(BaseModel):
id: int
code: str
fraction: Optional[str] = None
prefix: Optional[str] = None
type_code: Optional[str] = None
ad_valorem: Optional[float] = None
@@ -46,3 +48,36 @@ class USTariffFractionResponseDTO(BaseModel):
description: Optional[str] = None
created_at: datetime
updated_at: datetime
@model_validator(mode="before")
@classmethod
def format_code_and_fraction(cls, data: Any) -> Any:
# Check if data is an ORM model or dict
if hasattr(data, "code"):
raw_code = data.code
elif isinstance(data, dict):
raw_code = data.get("code")
else:
return data
if raw_code:
code_str = str(raw_code)
# fraction keeps the original formatted string
fraction = code_str
# code strips dots and hyphens
code = code_str.replace(".", "").replace("-", "")
if isinstance(data, dict):
data["code"] = code
data["fraction"] = fraction
else:
# If it's an ORM object, we can't easily modify the object's attribute
# cleanly without side effects for other things, so we convert it to dict
new_data = {
c.name: getattr(data, c.name) for c in data.__table__.columns
}
new_data["code"] = code
new_data["fraction"] = fraction
return new_data
return data

View File

@@ -17,10 +17,20 @@ 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",
id_name="id",
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 +54,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 +66,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)

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,
@@ -150,13 +98,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 +117,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 +145,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 +166,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