plantillas-proyectos/backend/api/v1/modules/sitar/README.md

SITAR API Module

Módulo reorganizado para integración con la API externa de SITAR (Sistema de Información de Aranceles).

Estructura

Cada recurso ahora tiene su propia carpeta con separación de responsabilidades:

backend/api/v1/modules/sitar/
├── common/                 # Servicio base compartido
│   ├── __init__.py
│   └── base_service.py     # SitarAPIBaseService con autenticación
├── tlcs/                   # TLCS (Tratados de Libre Comercio)
│   ├── __init__.py
│   ├── schemas.py          # TLCSResponse
│   ├── service.py          # TLCSService
│   └── router.py           # Endpoints FastAPI
├── fracciones/             # Fracciones arancelarias mexicanas
├── fracciones_usa/         # Fracciones USA
├── regulaciones/           # Regulaciones y restricciones
├── prosec/                 # PROSEC
├── precios_estimados/      # Precios estimados
├── fundamentos_tlc/        # Fundamentos TLC
├── aladi2/                 # ALADI2
├── cuotas2/                # Cuotas compensatorias
├── cupos/                  # Cupos de importación
├── fracciones_anteriores/  # Historial de fracciones
├── informacion_general/    # Información general
├── ieps/                   # IEPS
├── noms/                   # Normas Oficiales Mexicanas
├── rcg2/                   # Reglas de Carácter General
├── reit/                   # REIT
├── requisito_previo/       # Requisitos previos
├── vehiculos_marcas/       # Marcas de vehículos
├── vehiculos_modelos/      # Modelos de vehículos
├── __init__.py             # Exporta todos los servicios
└── main_router.py          # Router principal con todos los endpoints

Uso del Servicio

Desde cualquier parte del código

Cada recurso tiene su propio servicio singleton que maneja automáticamente la autenticación:

from api.v1.modules.sitar.tlcs import TLCSService
from api.v1.modules.sitar.fracciones import FraccionesService

# Obtener instancia singleton
tlcs_service = TLCSService.get_instance()

# Buscar TLCS (async)
tlcs_data = await tlcs_service.search(
    fraccion="84716001",
    pais="USA",
    limit=100
)

# Procesar resultados
for tlcs in tlcs_data:
    print(f"Tasa: {tlcs.TASATXT}, País: {tlcs.PAIS}")

Desde código síncrono

Si estás en contexto síncrono, usa asyncio.run():

import asyncio
from api.v1.modules.sitar.tlcs import TLCSService

tlcs_service = TLCSService.get_instance()
tlcs_data = asyncio.run(tlcs_service.search(fraccion="84716001", pais="USA"))

Métodos Disponibles

TLCS (Tratados de Libre Comercio)

• search_tlcs(fraccion, pais, nico, skip, limit) - Buscar TLCS
• get_tlcs_by_id(sysid, fraccion) - Obtener TLCS por ID

Fracciones Arancelarias

• search_fracciones(fraccion, nico, skip, limit) - Buscar fracciones mexicanas
• get_fraccion_by_id(sysid) - Obtener fracción por ID
• search_fracciones_usa(fraccion, skip, limit) - Buscar fracciones USA
• get_fraccion_usa_by_id(consecutivo) - Obtener fracción USA por ID
• search_fracciones_anteriores(fraccion_actual, fraccion_anterior, skip, limit) - Buscar historial de fracciones
• get_fracciones_anteriores_by_id(sysid) - Obtener historial por ID

Regulaciones y Restricciones

• search_regulaciones(fraccion, nico, skip, limit) - Buscar regulaciones
• get_regulacion_by_id(sysid) - Obtener regulación por ID
• search_noms(fraccion, pais, nico, skip, limit) - Buscar Normas Oficiales Mexicanas
• get_noms_by_id(sysid) - Obtener NOM por ID
• search_requisito_previo(fraccion, nico, skip, limit) - Buscar requisitos previos
• get_requisito_previo_by_id(sysid) - Obtener requisito previo por ID

PROSEC y Programas de Promoción

• search_prosec(fraccion, nico, skip, limit) - Buscar PROSEC
• get_prosec_by_id(sysid) - Obtener PROSEC por ID
• search_reit(fraccion, nico, skip, limit) - Buscar REIT
• get_reit_by_id(sysid) - Obtener REIT por ID

Precios e Impuestos

• search_precios_estimados(fraccion, nico, skip, limit) - Buscar precios estimados
• get_precio_estimado_by_id(sysid) - Obtener precio estimado por ID
• search_ieps(fraccion, nico, skip, limit) - Buscar IEPS
• get_ieps_by_id(consecutivo) - Obtener IEPS por ID

Fundamentos y Acuerdos

• search_fundamentos_tlc(fraccion, nico, tipat_only, skip, limit) - Buscar fundamentos TLC
• get_fundamento_tlc_by_id(sysid) - Obtener fundamento por ID
• search_aladi2(fraccion, pais, nico, skip, limit) - Buscar ALADI2
• get_aladi2_by_id(sysid) - Obtener ALADI2 por ID

Cuotas y Cupos

• search_cuotas2(fraccion, pais, nico, skip, limit) - Buscar cuotas compensatorias
• get_cuotas2_by_id(sysid) - Obtener cuota por ID
• search_cupos(fraccion, nico, skip, limit) - Buscar cupos de importación
• get_cupos_by_id(sysid) - Obtener cupo por ID

Reglas de Carácter General

• search_rcg2(fraccion, nico, skip, limit) - Buscar RCG2
• get_rcg2_by_id(sysid) - Obtener RCG2 por ID

Información General

• search_informacion_general(fraccion, nico, skip, limit) - Buscar información general
• get_informacion_general_by_id(sysid) - Obtener información general por ID

Vehículos

• search_vehiculos_marcas(fraccion, marca, skip, limit) - Buscar marcas de vehículos
• get_vehiculos_marcas_by_id(sysid) - Obtener marca por ID
• search_vehiculos_modelos(fraccion, marca, modelo, skip, limit) - Buscar modelos de vehículos
• get_vehiculos_modelos_by_id(sysid) - Obtener modelo por ID

Configuración

Requiere las siguientes variables de entorno:

SITAR_API_URL=https://api.sitar.example.com
SITAR_API_USER=your_username
SITAR_API_PASSWORD=your_password

Servicios Disponibles

Cada módulo sigue el mismo patrón con métodos search() y get_by_id():

TLCS - TLCSService

from api.v1.modules.sitar.tlcs import TLCSService
service = TLCSService.get_instance()
await service.search(fraccion="84716001", pais="USA", nico=None, skip=0, limit=100)
await service.get_by_id(sysid=123, fraccion="84716001")

Fracciones - FraccionesService

from api.v1.modules.sitar.fracciones import FraccionesService
service = FraccionesService.get_instance()
await service.search(fraccion="84716001", nico=None, skip=0, limit=100)
await service.get_by_id(sysid=123)

Fracciones USA - FraccionesUSAService

from api.v1.modules.sitar.fracciones_usa import FraccionesUSAService
service = FraccionesUSAService.get_instance()
await service.search(fraccion="84716001", skip=0, limit=100)
await service.get_by_id(consecutivo=123)

Otros servicios disponibles:

• RegulacionesService - Regulaciones y restricciones
• ProsecService - PROSEC (Programa de Promoción Sectorial)
• PreciosEstimadosService - Precios estimados
• FundamentosTLCService - Fundamentos de tratados de libre comercio
• Aladi2Service - ALADI2 (Asociación Latinoamericana de Integración)
• Cuotas2Service - Cuotas compensatorias
• CuposService - Cupos de importación
• FraccionesAnterioresService - Historial de fracciones
• InformacionGeneralService - Información general de fracciones
• IepsService - IEPS (Impuesto Especial sobre Producción y Servicios)
• NomsService - Normas Oficiales Mexicanas
• Rcg2Service - Reglas de Carácter General
• ReitService - Registro de Empresas de Industria Terminal
• RequisitoPrevioService - Requisitos previos
• VehiculosMarcasService - Marcas de vehículos
• VehiculosModelosService - Modelos de vehículos

```python
from fastapi import APIRouter, HTTPException
from api.v1.modules.sitar.tlcs import TLCSService

router = APIRouter()

@router.get("/tariff/{fraccion}")
async def get_tariff_info(fraccion: str, country: str = "USA"):
    """Endpoint que consulta información arancelaria"""
    try:
        tlcs_service = TLCSService.get_instance()
        data = await tlcs_service.search(fraccion=fraccion, pais=country)
        return {"success": True, "data": data}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Arquitectura

Servicio Base Compartido

Todos los servicios heredan de SitarAPIBaseService que maneja:

• Autenticación con token caching
• Renovación automática de tokens
• Manejo de errores HTTP
• Timeout configurable

Singleton Pattern

Cada servicio usa el patrón singleton para:

• Compartir la conexión HTTP
• Reutilizar tokens de autenticación
• Evitar múltiples instancias

Separación de Responsabilidades

• schemas.py: Modelos Pydantic para validación de datos
• service.py: Lógica de negocio y llamadas a API
• router.py: Endpoints FastAPI (opcional)
• init.py: Exportaciones públicas

Notas Importantes

1. Autenticación: El token se cachea y renueva automáticamente
2. Timeouts: Configurado a 10 segundos por defecto
3. Límites: Máximo 1000 registros por consulta
4. Errores: Todos los servicios lanzan excepciones httpx en caso de error
5. Async/Sync: Los servicios son async, usa asyncio.run() en código síncrono

Esto expondrá endpoints como:
- `GET /api/v1/sitar/tlcs/` - Buscar TLCS
- `GET /api/v1/sitar/fracciones/` - Buscar fracciones
- etc.

## Ejemplo Completo

```python
from api.v1.modules.sitar import SitarAPIService

async def get_tariff_info(fraccion: str, country: str):
    """Obtener información arancelaria completa"""
    sitar = SitarAPIService.get_instance()
  
    try:
        # Buscar TLCS
        tlcs = await sitar.search_tlcs(fraccion=fraccion, pais=country)
      
        if tlcs:
            first_tlcs = tlcs[0]
            return {
                "rate": first_tlcs.TASATXT,
                "adv_impo": float(first_tlcs.TASA1NUM or 0.0),
                "country": first_tlcs.PAIS
            }
      
        # Si no hay TLCS, buscar fracción general
        fracciones = await sitar.search_fracciones(fraccion=fraccion)
      
        if fracciones:
            first_frac = fracciones[0]
            return {
                "rate": first_frac.ADVIMPOTXT,
                "adv_impo": float(first_frac.ADVIMPONUM or 0.0),
                "description": first_frac.DESCRIPCION
            }
          
        return None
      
    except Exception as e:
        print(f"Error: {e}")
        return None

Manejo de Errores

El servicio lanza excepciones httpx.HTTPError en caso de error:

import httpx

try:
    tlcs = await sitar.search_tlcs(fraccion="invalid")
except httpx.HTTPStatusError as e:
    print(f"HTTP Error: {e.response.status_code}")
except httpx.RequestError as e:
    print(f"Request Error: {e}")
except Exception as e:
    print(f"General Error: {e}")