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

feature/api-sittar-actualizacion

Browse Source
This commit is contained in:
hreyes
2026-05-08 12:54:19 -06:00
parent 9bcbc5bc02
commit 315c94219b
16 changed files with 304 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
backend/api/v1/modules/a76/general_catalogs/fractions/tariff_fractions/service.py
View File

@@ -164,16 +164,13 @@ class TariffFractionService:
search_description = term
try:
usa_items = await usa_service.search(
usa_items, total = await usa_service.search_with_total(
fraccion=search_term,
descripcion=search_description,
skip=skip,
limit=limit,
)
items = [TariffFractionMapper.to_domain_usa(item) for item in usa_items]
total = len(items) + skip
if len(items) == limit:
total += 1
return items, total
except Exception as e:
import traceback
@@ -245,30 +242,25 @@ class TariffFractionService:
# Note: Sitar search might not return total count.
# We fetch page items. Pagination might be tricky if Sitar doesn't return total.
# Assuming Sitar returns a list.
sitar_items = await sitar_service.search(
sitar_items, total = await sitar_service.search_with_total(
fraccion=sitar_fraccion,
nico=sitar_nico,
description=sitar_description,
nivel=level_filter, # Dynamic level
skip=skip,
limit=limit
limit=limit,
)
# STRICT API USAGE:
# We do NOT fallback to local DB on empty list, as user requested strict API consumption.
# We also do NOT attempt enrichment as codes mismatch (API uses '010191A' vs Local '01012101').
# Map items
items = [TariffFractionMapper.to_domain(item) for item in sitar_items]
# Legacy browse behavior: keep table in ascending fracción order.
items = sorted(items, key=lambda row: ((row.code or ""), (row.nico or "")))
# Estimate total (Sitar service doesn't return total currently)
# If we got full limit, assume there are more.
total = len(items) + skip
if len(items) == limit:
total += 1 # Indicate more pages
# total comes from SITAR PaginatedFraccionesResponse (matches API-wide count for the query).
return items, total
except Exception as e:

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

@@ -103,7 +103,7 @@ async def list_us_tariff_fractions(
search_description = search
try:
sitar_items = await svc.search(
sitar_items, total = await svc.search_with_total(
fraccion=search_term,
descripcion=search_description,
skip=skip,
@@ -118,10 +118,6 @@ async def list_us_tariff_fractions(
"pages": 0,
}
total = len(sitar_items) + skip
if len(sitar_items) == page_size:
total += 1
items = [
USTariffFractionResponseDTO.model_validate(_sitar_row_to_us_response_payload(row))
for row in sitar_items

5
backend/api/v1/modules/core/tasks_tracking/dispatch.py
View File

@@ -3,7 +3,7 @@ from typing import Any
from celery import Task
from sqlalchemy.orm import Session
from core.database import rls_company_var, rls_tenant_var
from core.database import reset_rls_context_tokens, rls_company_var, rls_tenant_var
from .service import TaskTrackerService
@@ -39,8 +39,7 @@ def track_and_dispatch(
headers=headers,
)
finally:
rls_tenant_var.reset(token_t)
rls_company_var.reset(token_c)
reset_rls_context_tokens(token_t, token_c)
tracker = TaskTrackerService(db)
tracker.register_dispatch(

25
backend/api/v1/modules/sitar/common/base_service.py
View File

@@ -6,7 +6,7 @@ All specific resource services inherit from this.
"""
import os
from typing import Optional, Dict, Any
from typing import Optional, Dict, Any, List, Tuple
from datetime import datetime, timedelta
import httpx
@@ -17,6 +17,29 @@ class SitarAPIBaseService:
_token: Optional[str] = None
_token_expires: Optional[datetime] = None
@staticmethod
def _parse_paginated_list_and_total(data: Any) -> Tuple[List[Any], int]:
"""
Paginated list endpoints (fracciones, fracciones-usa) return:
{ "data": [...], "total", "page", "limit", "total_pages" }.
Older responses may be a plain JSON array; then total is len(rows) for that page only.
"""
if isinstance(data, list):
return data, len(data)
if isinstance(data, dict) and isinstance(data.get("data"), list):
rows = data["data"]
raw_total = data.get("total")
total = int(raw_total) if raw_total is not None else len(rows)
return rows, total
raise ValueError(
f"Unexpected SITAR list response shape: {type(data).__name__}"
)
@staticmethod
def _unwrap_paginated_list(data: Any) -> List[Any]:
rows, _ = SitarAPIBaseService._parse_paginated_list_and_total(data)
return rows
def __init__(self):
"""Initialize base service with API credentials"""
self.base_url = os.getenv("SITAR_API_URL")

33
backend/api/v1/modules/sitar/fracciones/service.py
View File

@@ -1,7 +1,7 @@
"""Fracciones Service"""
import asyncio
from typing import Optional, List
from typing import Optional, List, Tuple
from ..common import SitarAPIBaseService
from .schemas import FraccionesResponse
@@ -18,7 +18,7 @@ class FraccionesService(SitarAPIBaseService):
cls._instance = cls()
return cls._instance
async def search(
async def search_with_total(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
@@ -26,8 +26,8 @@ class FraccionesService(SitarAPIBaseService):
nivel: Optional[int] = None,
skip: int = 0,
limit: int = 100,
) -> List[FraccionesResponse]:
"""Search Mexican tariff fractions"""
) -> Tuple[List[FraccionesResponse], int]:
"""Search Mexican tariff fractions; total matches SITAR PaginatedFraccionesResponse.total."""
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
@@ -38,8 +38,29 @@ class FraccionesService(SitarAPIBaseService):
if nivel is not None:
params["nivel"] = nivel
data = await self._make_request("GET", "/api/v1/fracciones/", params=params)
return [FraccionesResponse(**item) for item in data]
raw = await self._make_request("GET", "/api/v1/fracciones/", params=params)
rows, total = self._parse_paginated_list_and_total(raw)
return [FraccionesResponse(**item) for item in rows], total
async def search(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
description: Optional[str] = None,
nivel: Optional[int] = None,
skip: int = 0,
limit: int = 100,
) -> List[FraccionesResponse]:
"""Search Mexican tariff fractions"""
items, _ = await self.search_with_total(
fraccion=fraccion,
nico=nico,
description=description,
nivel=nivel,
skip=skip,
limit=limit,
)
return items
async def get_by_id(self, sysid: int) -> FraccionesResponse:
"""Get single Fraccion record by SYSID"""

42
backend/api/v1/modules/sitar/fracciones_usa/service.py
View File

@@ -1,7 +1,7 @@
"""Fracciones USA Service"""
import logging
from typing import Optional, List
from typing import Optional, List, Tuple
from ..common import SitarAPIBaseService
from .schemas import FraccionesUSAResponse
@@ -22,6 +22,24 @@ class FraccionesUSAService(SitarAPIBaseService):
cls._instance = cls()
return cls._instance
async def search_with_total(
self,
fraccion: Optional[str] = None,
descripcion: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> Tuple[List[FraccionesUSAResponse], int]:
"""Search USA tariff fractions; total matches SITAR PaginatedFraccionesUSAResponse.total."""
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if descripcion:
params["descripcion"] = descripcion
raw = await self._make_request("GET", "api/v1/fracciones-usa/", params=params)
rows, total = self._parse_paginated_list_and_total(raw)
return [FraccionesUSAResponse(**item) for item in rows], total
async def search(
self,
fraccion: Optional[str] = None,
@@ -30,14 +48,13 @@ class FraccionesUSAService(SitarAPIBaseService):
limit: int = 100,
) -> List[FraccionesUSAResponse]:
"""Search USA tariff fractions"""
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if descripcion:
params["descripcion"] = descripcion
data = await self._make_request("GET", "api/v1/fracciones-usa/", params=params)
return [FraccionesUSAResponse(**item) for item in data]
items, _ = await self.search_with_total(
fraccion=fraccion,
descripcion=descripcion,
skip=skip,
limit=limit,
)
return items
async def get_by_id(self, consecutivo: int) -> FraccionesUSAResponse:
"""Get single USA Fraccion record by CONSECUTIVO"""
@@ -68,10 +85,9 @@ class FraccionesUSAService(SitarAPIBaseService):
params["descripcion"] = descripcion
try:
data = service._make_request_sync("GET", "api/v1/fracciones-usa/", params=params)
if not isinstance(data, list):
return []
return [FraccionesUSAResponse(**item) for item in data]
raw = service._make_request_sync("GET", "api/v1/fracciones-usa/", params=params)
rows, _ = service._parse_paginated_list_and_total(raw)
return [FraccionesUSAResponse(**item) for item in rows]
except Exception as exc:
logger.warning("SITAR fracciones-usa search_sync failed: %s", exc)
return []

10
backend/api/v1/modules/sitar/openapi/README.md Normal file
View File

@@ -0,0 +1,10 @@
# SITAR API — especificación OpenAPI rescatada
| Campo | Valor |
|-------|--------|
| Fuente | `GET http://api.sitar.aduanasoft.com:880/fractions/api/v1/openapi.json` |
| UI Swagger | `http://api.sitar.aduanasoft.com:880/fractions/docs` |
| Rescatado | 2026-05-08 |
| Archivo | [sitar-api-openapi-3.1.json](sitar-api-openapi-3.1.json) (OpenAPI 3.1, «SITAR API - DEV») |
Las rutas documentadas usan prefijo `/api/v1/...`. Las peticiones desde este proyecto anteponen `{SITAR_API_URL}/fractions/` (ver `SitarAPIBaseService`), por ejemplo: `{SITAR_API_URL}/fractions/api/v1/tlcs/`.

1
backend/api/v1/modules/sitar/openapi/sitar-api-openapi-3.1.json Normal file
View File

File diff suppressed because one or more lines are too long

19
backend/api/v1/modules/sitar/prosec/schemas.py
View File

@@ -1,19 +1,20 @@
"""PROSEC Schemas"""
"""PROSEC Schemas — aligned with SITAR OpenAPI ProsecResponse."""
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field
class ProsecResponse(BaseModel):
"""PROSEC (Programa de Promoción Sectorial)"""
"""PROSEC (Programa de Promoción Sectorial)."""
FRACCION: Optional[str] = Field(None, max_length=10)
PRODUCTO: Optional[str] = Field(None, max_length=999)
TASA: Optional[str] = Field(None, max_length=19)
SECTOR: Optional[str] = Field(None, max_length=2)
ANEXO: Optional[str] = Field(None, max_length=19)
DOF: Optional[str] = Field(None, max_length=8)
NOTAS: Optional[str] = Field(None, max_length=5000)
FRACCION: Optional[str] = Field("", max_length=10)
ARTICULO: Optional[str] = Field("", max_length=3)
SECTOR: Optional[str] = Field("", max_length=6)
TASATXT: Optional[str] = Field("", max_length=19)
TASANUM: Optional[str] = Field(default="0")
TIPOTASA: Optional[int] = Field(default=0)
DOF: Optional[str] = Field("", max_length=8)
OBSERVACION: Optional[str] = ""
NICO: Optional[str] = Field("", max_length=2)
SYSID: int

22
backend/api/v1/modules/sitar/tlcs/schemas.py
View File

@@ -1,21 +1,23 @@
"""TLCS Schemas"""
"""TLCS Schemas — aligned with SITAR OpenAPI TLCSResponse."""
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field
class TLCSResponse(BaseModel):
"""TLCS (Tratados de Libre Comercio) response model"""
"""TLCS (Tratados de Libre Comercio) response model."""
FRACCION: Optional[str] = Field(None, max_length=8)
PAIS: Optional[str] = Field(None, max_length=3)
TASATXT: Optional[str] = Field(None, max_length=19)
TASANUM: Optional[str] = None
TASACALCULADA: Optional[str] = Field(None, max_length=19)
TLC: Optional[str] = Field(None, max_length=6)
NOTA: Optional[str] = None
FRACCION: str = Field(max_length=10)
PAIS: str = Field(max_length=3)
ORDEN: Optional[int] = None
TASATXT: Optional[str] = Field(None, max_length=44)
TIPOTASA: Optional[int] = None
TASA1NUM: Optional[str] = None
FACTOR1: Optional[str] = None
TASA2NUM: Optional[str] = None
FACTOR2: Optional[str] = None
DOF: Optional[str] = Field(None, max_length=8)
OBSERVACION: Optional[str] = None
NOTAS: Optional[str] = Field(None, max_length=999)
NICO: Optional[str] = Field("", max_length=2)
SYSID: int