plantillas-proyectos/backend/api/v1/modules/sitar

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}")