feat: Implement SITAR modules for PROSEC, RCG2, Regulaciones, REIT, RequisitoPrevio, TLCS, and Vehiculos

- Added schemas, services, and routers for PROSEC, RCG2, Regulaciones, REIT, RequisitoPrevio, TLCS, and Vehiculos modules.
- Each module includes search and get by ID functionalities.
- Integrated FastAPI routers for each module into the main SITAR router.
- Ensured proper response models using Pydantic for data validation.
This commit is contained in:
2026-02-04 23:28:46 -06:00
parent d4014df5d5
commit f38c13f851
81 changed files with 2580 additions and 0 deletions

View File

@@ -0,0 +1,320 @@
# 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:
```python
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()`:
```python
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:
```bash
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`
```python
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`
```python
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`
```python
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
```### Desde endpoints asíncronos
```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:
```python
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}")
```

View File

@@ -0,0 +1,90 @@
"""
SITAR API Module
Módulo reorganizado para integración con la API externa de SITAR.
Cada recurso tiene su propia carpeta con schemas, service y router.
Estructura:
backend/api/v1/modules/sitar/
├── common/ # Base service compartido
├── tlcs/ # TLCS (Tratados de Libre Comercio)
├── fracciones/ # Fracciones arancelarias mexicanas
├── fracciones_usa/ # Fracciones arancelarias 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
Uso:
# Importar servicios específicos
from api.v1.modules.sitar.tlcs import TLCSService
from api.v1.modules.sitar.fracciones import FraccionesService
# Usar servicios
tlcs_service = TLCSService.get_instance()
data = await tlcs_service.search(fraccion="84716001")
# Importar routers para FastAPI
from api.v1.modules.sitar.tlcs import router as tlcs_router
from api.v1.modules.sitar.fracciones import router as fracciones_router
app.include_router(tlcs_router, prefix="/api/v1/sitar/tlcs", tags=["sitar-tlcs"])
"""
from .common import SitarAPIBaseService
# Import all services for convenient access
from .tlcs import TLCSService
from .fracciones import FraccionesService
from .fracciones_usa import FraccionesUSAService
from .regulaciones import RegulacionesService
from .prosec import ProsecService
from .precios_estimados import PreciosEstimadosService
from .fundamentos_tlc import FundamentosTLCService
from .aladi2 import Aladi2Service
from .cuotas2 import Cuotas2Service
from .cupos import CuposService
from .fracciones_anteriores import FraccionesAnterioresService
from .informacion_general import InformacionGeneralService
from .ieps import IepsService
from .noms import NomsService
from .rcg2 import Rcg2Service
from .reit import ReitService
from .requisito_previo import RequisitoPrevioService
from .vehiculos_marcas import VehiculosMarcasService
from .vehiculos_modelos import VehiculosModelosService
__all__ = [
"SitarAPIBaseService",
"TLCSService",
"FraccionesService",
"FraccionesUSAService",
"RegulacionesService",
"ProsecService",
"PreciosEstimadosService",
"FundamentosTLCService",
"Aladi2Service",
"Cuotas2Service",
"CuposService",
"FraccionesAnterioresService",
"InformacionGeneralService",
"IepsService",
"NomsService",
"Rcg2Service",
"ReitService",
"RequisitoPrevioService",
"VehiculosMarcasService",
"VehiculosModelosService",
]

View File

@@ -0,0 +1,7 @@
"""Aladi2 Module"""
from .schemas import Aladi2Response
from .service import Aladi2Service
from .router import router
__all__ = ["Aladi2Response", "Aladi2Service", "router"]

View File

@@ -0,0 +1,37 @@
"""Aladi2 Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query
from fastapi import Depends
from core.security import get_current_user
from .service import Aladi2Service
from .schemas import Aladi2Response
router = APIRouter()
@router.get("/", response_model=List[Aladi2Response])
async def search(
fraccion: Optional[str] = Query(None),
pais: Optional[str] = Query(None),
nico: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = Aladi2Service.get_instance()
return await service.search(
fraccion=fraccion, pais=pais, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=Aladi2Response)
async def get_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
return await Aladi2Service.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,24 @@
"""ALADI2 Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class Aladi2Response(BaseModel):
"""ALADI2 (Asociación Latinoamericana de Integración)"""
FRACCION: Optional[str] = Field(None, max_length=10)
ACUERDO: Optional[str] = Field(None, max_length=49)
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)
TIPOTASA: Optional[int] = None
DOF: Optional[str] = Field(None, max_length=8)
NOTAS: Optional[str] = Field(None, max_length=499)
OBSERVACIONES: Optional[str] = None
NICO: Optional[str] = Field(None, max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,37 @@
"""Aladi2 Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import Aladi2Response
class Aladi2Service(SitarAPIBaseService):
_instance: Optional["Aladi2Service"] = None
@classmethod
def get_instance(cls) -> "Aladi2Service":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
pais: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[Aladi2Response]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if pais:
params["pais"] = pais
if nico:
params["nico"] = nico
data = await self._make_request("GET", "/api/v1/aladi2/", params=params)
return [Aladi2Response(**item) for item in data]
async def get_by_id(self, sysid: int) -> Aladi2Response:
data = await self._make_request("GET", f"/api/v1/aladi2/{sysid}")
return Aladi2Response(**data)

View File

@@ -0,0 +1,5 @@
"""SITAR API Common Module"""
from .base_service import SitarAPIBaseService
__all__ = ["SitarAPIBaseService"]

View File

@@ -0,0 +1,104 @@
"""
SITAR API Base Service
Base service class for SITAR API authentication and HTTP requests.
All specific resource services inherit from this.
"""
import os
from typing import Optional, Dict, Any
from datetime import datetime, timedelta
import httpx
class SitarAPIBaseService:
"""Base service for SITAR API integration with authentication"""
_token: Optional[str] = None
_token_expires: Optional[datetime] = None
def __init__(self):
"""Initialize base service with API credentials"""
self.base_url = os.getenv("SITAR_API_URL")
self.username = os.getenv("SITAR_API_USER")
self.password = os.getenv("SITAR_API_PASSWORD")
self.timeout = 10.0
if not all([self.base_url, self.username, self.password]):
raise ValueError(
"Missing SITAR API configuration. "
"Set SITAR_API_URL, SITAR_API_USER, and SITAR_API_PASSWORD environment variables."
)
async def _get_token(self) -> str:
"""
Get authentication token, refreshing if necessary
Returns:
str: Bearer token for API authentication
Raises:
httpx.HTTPError: If authentication fails
"""
# Return cached token if still valid
if self._token and self._token_expires and datetime.now() < self._token_expires:
return self._token
# Authenticate and get new token
login_url = f"{self.base_url}/fractions/api/v1/auth/login"
payload = {"username": self.username, "password": self.password}
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.post(login_url, json=payload)
response.raise_for_status()
data = response.json()
self._token = data.get("access_token") or data.get("token")
if not self._token:
raise ValueError("No token received from SITAR API")
# Set token expiration (assume 1 hour if not specified)
self._token_expires = datetime.now() + timedelta(hours=1)
return self._token
async def _make_request(
self,
method: str,
endpoint: str,
params: Optional[Dict[str, Any]] = None,
json_data: Optional[Dict[str, Any]] = None,
) -> Any:
"""
Make authenticated request to SITAR API
Args:
method: HTTP method (GET, POST, etc.)
endpoint: API endpoint path
params: Query parameters
json_data: JSON body data
Returns:
JSON response data
Raises:
httpx.HTTPError: If request fails
"""
token = await self._get_token()
url = f"{self.base_url}/fractions/{endpoint}"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
}
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.request(
method=method,
url=url,
params=params,
json=json_data,
headers=headers,
)
response.raise_for_status()
return response.json()

View File

@@ -0,0 +1,7 @@
"""Cuotas2 Module"""
from .schemas import Cuotas2Response
from .service import Cuotas2Service
from .router import router
__all__ = ["Cuotas2Response", "Cuotas2Service", "router"]

View File

@@ -0,0 +1,35 @@
"""Cuotas2 Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import Cuotas2Service
from .schemas import Cuotas2Response
router = APIRouter()
@router.get("/", response_model=List[Cuotas2Response])
async def search(
fraccion: Optional[str] = Query(None),
pais: Optional[str] = Query(None),
nico: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = Cuotas2Service.get_instance()
return await service.search(
fraccion=fraccion, pais=pais, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=Cuotas2Response)
async def get_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
return await Cuotas2Service.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,30 @@
"""Cuotas2 Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class Cuotas2Response(BaseModel):
"""Cuotas compensatorias"""
FRACCION: Optional[str] = Field(None, max_length=10)
PRODUCTO: Optional[str] = Field(None, max_length=300)
PAIS: Optional[str] = Field(None, max_length=3)
DOF: Optional[str] = Field(None, max_length=8)
DOF2: Optional[str] = Field(None, max_length=149)
EMPRESA: Optional[str] = Field(None, max_length=100)
INCOTERM: Optional[str] = Field(None, max_length=10)
DOLARES: Optional[str] = None
CLAVEMONEDA: Optional[str] = Field(None, max_length=3)
CLAVEUM: Optional[str] = Field(None, max_length=2)
PORCENTAJE: Optional[str] = None
DESCRIPCIONDOLARES: Optional[str] = Field(None, max_length=50)
NOTAS: Optional[str] = None
ESTATUS: Optional[str] = Field(None, max_length=19)
NUMERORESOLUCION: Optional[int] = None
RESOLUCION: Optional[str] = Field(None, max_length=150)
NICO: Optional[str] = Field("", max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,37 @@
"""Cuotas2 Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import Cuotas2Response
class Cuotas2Service(SitarAPIBaseService):
_instance: Optional["Cuotas2Service"] = None
@classmethod
def get_instance(cls) -> "Cuotas2Service":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
pais: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[Cuotas2Response]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if pais:
params["pais"] = pais
if nico:
params["nico"] = nico
data = await self._make_request("GET", "/api/v1/cuotas2/", params=params)
return [Cuotas2Response(**item) for item in data]
async def get_by_id(self, sysid: int) -> Cuotas2Response:
data = await self._make_request("GET", f"/api/v1/cuotas2/{sysid}")
return Cuotas2Response(**data)

View File

@@ -0,0 +1,7 @@
"""Cupos Module"""
from .schemas import CuposResponse
from .service import CuposService
from .router import router
__all__ = ["CuposResponse", "CuposService", "router"]

View File

@@ -0,0 +1,34 @@
"""Cupos Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import CuposService
from .schemas import CuposResponse
router = APIRouter()
@router.get("/", response_model=List[CuposResponse])
async def search(
fraccion: Optional[str] = Query(None),
nico: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = CuposService.get_instance()
return await service.search(
fraccion=fraccion, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=CuposResponse)
async def get_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
return await CuposService.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,25 @@
"""Cupos Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class CuposResponse(BaseModel):
"""Cupos de importación"""
FRACCION: Optional[str] = Field(None, max_length=10)
NUMEROCUPO: Optional[int] = None
DOF: Optional[str] = Field(None, max_length=8)
PERMISO: Optional[str] = Field(None, max_length=2)
CONDICION: Optional[str] = Field(None, max_length=199)
VIGENCIA: Optional[str] = Field(None, max_length=8)
PAIS: Optional[str] = Field(None, max_length=5)
OBSERVACIONES: Optional[str] = Field(None, max_length=5000)
ADVIMPO: Optional[str] = Field(None, max_length=20)
ADVEXPO: Optional[str] = Field(None, max_length=20)
TIPOOPERACION: Optional[int] = None
NICO: Optional[str] = Field("", max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,34 @@
"""Cupos Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import CuposResponse
class CuposService(SitarAPIBaseService):
_instance: Optional["CuposService"] = None
@classmethod
def get_instance(cls) -> "CuposService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[CuposResponse]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if nico:
params["nico"] = nico
data = await self._make_request("GET", "/api/v1/cupos/", params=params)
return [CuposResponse(**item) for item in data]
async def get_by_id(self, sysid: int) -> CuposResponse:
data = await self._make_request("GET", f"/api/v1/cupos/{sysid}")
return CuposResponse(**data)

View File

@@ -0,0 +1,7 @@
"""Fracciones Module"""
from .schemas import FraccionesResponse
from .service import FraccionesService
from .router import router
__all__ = ["FraccionesResponse", "FraccionesService", "router"]

View File

@@ -0,0 +1,43 @@
"""Fracciones Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import FraccionesService
from .schemas import FraccionesResponse
router = APIRouter()
@router.get("/", response_model=List[FraccionesResponse])
async def search_fracciones(
fraccion: Optional[str] = Query(None, description="Fracción arancelaria"),
nico: Optional[str] = Query(None, description="NICO"),
skip: int = Query(0, ge=0, description="Registros a saltar"),
limit: int = Query(100, ge=1, le=1000, description="Máximo de registros"),
current_user: dict = Depends(get_current_user),
):
"""Search Mexican tariff fractions"""
try:
service = FraccionesService.get_instance()
return await service.search(
fraccion=fraccion, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Error fetching Fracciones data: {str(e)}"
)
@router.get("/{sysid}", response_model=FraccionesResponse)
async def get_fraccion_by_id(
sysid: int, current_user: dict = Depends(get_current_user)
):
"""Get single Fraccion record by SYSID"""
try:
service = FraccionesService.get_instance()
return await service.get_by_id(sysid)
except Exception as e:
raise HTTPException(
status_code=404, detail=f"Fraccion record not found: {str(e)}"
)

View File

@@ -0,0 +1,20 @@
"""Fracciones Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class FraccionesResponse(BaseModel):
"""Fracciones arancelarias mexicanas"""
FRACCION: Optional[str] = Field(None, max_length=10)
DESCRIPCION: Optional[str] = None
UMT: Optional[str] = Field(None, max_length=2)
UMC: Optional[str] = Field(None, max_length=2)
IMPUESTO_EXP: Optional[str] = Field(None, max_length=19)
IMPUESTO_IMP: Optional[str] = Field(None, max_length=19)
NICO: Optional[str] = Field("", max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,40 @@
"""Fracciones Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import FraccionesResponse
class FraccionesService(SitarAPIBaseService):
"""Service for Fracciones operations"""
_instance: Optional["FraccionesService"] = None
@classmethod
def get_instance(cls) -> "FraccionesService":
"""Get singleton instance"""
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[FraccionesResponse]:
"""Search Mexican tariff fractions"""
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if nico:
params["nico"] = nico
data = await self._make_request("GET", "/api/v1/fracciones/", params=params)
return [FraccionesResponse(**item) for item in data]
async def get_by_id(self, sysid: int) -> FraccionesResponse:
"""Get single Fraccion record by SYSID"""
data = await self._make_request("GET", f"/api/v1/fracciones/{sysid}")
return FraccionesResponse(**data)

View File

@@ -0,0 +1,7 @@
"""FraccionesAnteriores Module"""
from .schemas import FraccionesAnterioresResponse
from .service import FraccionesAnterioresService
from .router import router
__all__ = ["FraccionesAnterioresResponse", "FraccionesAnterioresService", "router"]

View File

@@ -0,0 +1,37 @@
"""FraccionesAnteriores Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import FraccionesAnterioresService
from .schemas import FraccionesAnterioresResponse
router = APIRouter()
@router.get("/", response_model=List[FraccionesAnterioresResponse])
async def search(
fraccion_actual: Optional[str] = Query(None),
fraccion_anterior: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = FraccionesAnterioresService.get_instance()
return await service.search(
fraccion_actual=fraccion_actual,
fraccion_anterior=fraccion_anterior,
skip=skip,
limit=limit,
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=FraccionesAnterioresResponse)
async def get_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
return await FraccionesAnterioresService.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,15 @@
"""Fracciones Anteriores Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class FraccionesAnterioresResponse(BaseModel):
"""Fracciones arancelarias anteriores (histórico)"""
FRACCIONACTUAL: Optional[str] = Field(None, max_length=10)
FRACCIONANTERIOR: Optional[str] = Field(None, max_length=10)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,36 @@
"""FraccionesAnteriores Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import FraccionesAnterioresResponse
class FraccionesAnterioresService(SitarAPIBaseService):
_instance: Optional["FraccionesAnterioresService"] = None
@classmethod
def get_instance(cls) -> "FraccionesAnterioresService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion_actual: Optional[str] = None,
fraccion_anterior: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[FraccionesAnterioresResponse]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion_actual:
params["fraccion_actual"] = fraccion_actual
if fraccion_anterior:
params["fraccion_anterior"] = fraccion_anterior
data = await self._make_request(
"GET", "/api/v1/fracciones-anteriores/", params=params
)
return [FraccionesAnterioresResponse(**item) for item in data]
async def get_by_id(self, sysid: int) -> FraccionesAnterioresResponse:
data = await self._make_request("GET", f"/api/v1/fracciones-anteriores/{sysid}")
return FraccionesAnterioresResponse(**data)

View File

@@ -0,0 +1,7 @@
"""Fracciones USA Module"""
from .schemas import FraccionesUSAResponse
from .service import FraccionesUSAService
from .router import router
__all__ = ["FraccionesUSAResponse", "FraccionesUSAService", "router"]

View File

@@ -0,0 +1,41 @@
"""Fracciones USA Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import FraccionesUSAService
from .schemas import FraccionesUSAResponse
router = APIRouter()
@router.get("/", response_model=List[FraccionesUSAResponse])
async def search_fracciones_usa(
fraccion: Optional[str] = Query(None, description="Fracción arancelaria USA"),
skip: int = Query(0, ge=0, description="Registros a saltar"),
limit: int = Query(100, ge=1, le=1000, description="Máximo de registros"),
current_user: dict = Depends(get_current_user),
):
"""Search USA tariff fractions"""
try:
service = FraccionesUSAService.get_instance()
return await service.search(fraccion=fraccion, skip=skip, limit=limit)
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Error fetching USA Fracciones data: {str(e)}"
)
@router.get("/{consecutivo}", response_model=FraccionesUSAResponse)
async def get_fraccion_usa_by_id(
consecutivo: int,
current_user: dict = Depends(get_current_user),
):
"""Get single USA Fraccion record by CONSECUTIVO"""
try:
service = FraccionesUSAService.get_instance()
return await service.get_by_id(consecutivo)
except Exception as e:
raise HTTPException(
status_code=404, detail=f"USA Fraccion record not found: {str(e)}"
)

View File

@@ -0,0 +1,16 @@
"""Fracciones USA Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class FraccionesUSAResponse(BaseModel):
"""USA tariff fractions"""
FRACCIONUSA: Optional[str] = Field(None, max_length=10)
DESCRIPCIONUSA: Optional[str] = None
CLAVEUM: Optional[str] = Field(None, max_length=2)
CONSECUTIVO: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,37 @@
"""Fracciones USA Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import FraccionesUSAResponse
class FraccionesUSAService(SitarAPIBaseService):
"""Service for USA Fracciones operations"""
_instance: Optional["FraccionesUSAService"] = None
@classmethod
def get_instance(cls) -> "FraccionesUSAService":
"""Get singleton instance"""
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[FraccionesUSAResponse]:
"""Search USA tariff fractions"""
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
data = await self._make_request("GET", "/api/v1/fracciones-usa/", params=params)
return [FraccionesUSAResponse(**item) for item in data]
async def get_by_id(self, consecutivo: int) -> FraccionesUSAResponse:
"""Get single USA Fraccion record by CONSECUTIVO"""
data = await self._make_request("GET", f"/api/v1/fracciones-usa/{consecutivo}")
return FraccionesUSAResponse(**data)

View File

@@ -0,0 +1,7 @@
"""Fundamentos TLC Module"""
from .schemas import FundamentosTLCResponse
from .service import FundamentosTLCService
from .router import router
__all__ = ["FundamentosTLCResponse", "FundamentosTLCService", "router"]

View File

@@ -0,0 +1,37 @@
"""Fundamentos TLC Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import FundamentosTLCService
from .schemas import FundamentosTLCResponse
router = APIRouter()
@router.get("/", response_model=List[FundamentosTLCResponse])
async def search_fundamentos(
fraccion: Optional[str] = Query(None),
nico: Optional[str] = Query(None),
tipat_only: bool = Query(False),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = FundamentosTLCService.get_instance()
return await service.search(
fraccion=fraccion, nico=nico, tipat_only=tipat_only, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=FundamentosTLCResponse)
async def get_fundamento_by_id(
sysid: int, current_user: dict = Depends(get_current_user)
):
try:
return await FundamentosTLCService.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,27 @@
"""Fundamentos TLC Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class FundamentosTLCResponse(BaseModel):
"""Fundamentos de tratados de libre comercio"""
FRACCION: Optional[str] = Field(None, max_length=8)
PAIS: Optional[str] = Field(None, max_length=3)
ACUERDO: Optional[str] = Field(None, max_length=5)
TASA1TXT: Optional[str] = Field(None, max_length=19)
TASA1NUM: Optional[str] = None
TASA2TXT: Optional[str] = Field(None, max_length=19)
TASA2NUM: Optional[str] = None
FUNDAMENTO1: Optional[str] = None
FUNDAMENTO2: Optional[str] = None
PERIODO: Optional[str] = Field(None, max_length=299)
MODALIDAD: Optional[str] = Field(None, max_length=999)
DOF: Optional[str] = Field(None, max_length=8)
NOTAS: Optional[str] = Field(None, max_length=20)
NICO: Optional[str] = Field("", max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,40 @@
"""Fundamentos TLC Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import FundamentosTLCResponse
class FundamentosTLCService(SitarAPIBaseService):
_instance: Optional["FundamentosTLCService"] = None
@classmethod
def get_instance(cls) -> "FundamentosTLCService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
tipat_only: bool = False,
skip: int = 0,
limit: int = 100,
) -> List[FundamentosTLCResponse]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if nico:
params["nico"] = nico
endpoint = (
"/api/v1/fundamentos-tlc/tipat"
if tipat_only
else "/api/v1/fundamentos-tlc/"
)
data = await self._make_request("GET", endpoint, params=params)
return [FundamentosTLCResponse(**item) for item in data]
async def get_by_id(self, sysid: int) -> FundamentosTLCResponse:
data = await self._make_request("GET", f"/api/v1/fundamentos-tlc/{sysid}")
return FundamentosTLCResponse(**data)

View File

@@ -0,0 +1,7 @@
"""Ieps Module"""
from .schemas import IepsResponse
from .service import IepsService
from .router import router
__all__ = ["IepsResponse", "IepsService", "router"]

View File

@@ -0,0 +1,34 @@
"""Ieps Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import IepsService
from .schemas import IepsResponse
router = APIRouter()
@router.get("/", response_model=List[IepsResponse])
async def search(
fraccion: Optional[str] = Query(None),
nico: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = IepsService.get_instance()
return await service.search(
fraccion=fraccion, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{consecutivo}", response_model=IepsResponse)
async def get_by_id(consecutivo: int, current_user: dict = Depends(get_current_user)):
try:
return await IepsService.get_instance().get_by_id(consecutivo)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,20 @@
"""IEPS Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class IepsResponse(BaseModel):
"""IEPS (Impuesto Especial sobre Producción y Servicios)"""
FRACCION_SIN_PUNTO: Optional[str] = Field(None, max_length=10)
FRACCION_CON_PUNTO: Optional[str] = Field(None, max_length=10)
FUNDAMENTO: Optional[str] = Field(None, max_length=1000)
CONDICION: Optional[str] = Field(None, max_length=5000)
TASA: Optional[str] = None
TASAESPECIFICO: Optional[str] = None
NICO: Optional[str] = Field("", max_length=2)
CONSECUTIVO: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,34 @@
"""Ieps Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import IepsResponse
class IepsService(SitarAPIBaseService):
_instance: Optional["IepsService"] = None
@classmethod
def get_instance(cls) -> "IepsService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[IepsResponse]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if nico:
params["nico"] = nico
data = await self._make_request("GET", "/api/v1/ieps/", params=params)
return [IepsResponse(**item) for item in data]
async def get_by_id(self, consecutivo: int) -> IepsResponse:
data = await self._make_request("GET", f"/api/v1/ieps/{consecutivo}")
return IepsResponse(**data)

View File

@@ -0,0 +1,7 @@
"""InformacionGeneral Module"""
from .schemas import InformacionGeneralResponse
from .service import InformacionGeneralService
from .router import router
__all__ = ["InformacionGeneralResponse", "InformacionGeneralService", "router"]

View File

@@ -0,0 +1,34 @@
"""InformacionGeneral Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import InformacionGeneralService
from .schemas import InformacionGeneralResponse
router = APIRouter()
@router.get("/", response_model=List[InformacionGeneralResponse])
async def search(
fraccion: Optional[str] = Query(None),
nico: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = InformacionGeneralService.get_instance()
return await service.search(
fraccion=fraccion, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=InformacionGeneralResponse)
async def get_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
return await InformacionGeneralService.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,21 @@
"""Información General Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class InformacionGeneralResponse(BaseModel):
"""Información general de fracciones"""
OBSERVACIONES: Optional[str] = Field(None, max_length=1000)
NOTAOBSERVACIONES: Optional[str] = Field(None, max_length=1000)
FRACCION: Optional[str] = Field(None, max_length=10)
TIPOOPERACION: Optional[int] = None
ORDEN: Optional[int] = None
LEYENDA: Optional[str] = None
NOTA: Optional[str] = None
NICO: Optional[str] = Field("", max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,36 @@
"""InformacionGeneral Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import InformacionGeneralResponse
class InformacionGeneralService(SitarAPIBaseService):
_instance: Optional["InformacionGeneralService"] = None
@classmethod
def get_instance(cls) -> "InformacionGeneralService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[InformacionGeneralResponse]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if nico:
params["nico"] = nico
data = await self._make_request(
"GET", "/api/v1/informacion-general/", params=params
)
return [InformacionGeneralResponse(**item) for item in data]
async def get_by_id(self, sysid: int) -> InformacionGeneralResponse:
data = await self._make_request("GET", f"/api/v1/informacion-general/{sysid}")
return InformacionGeneralResponse(**data)

View File

@@ -0,0 +1,7 @@
"""Noms Module"""
from .schemas import NomsResponse
from .service import NomsService
from .router import router
__all__ = ["NomsResponse", "NomsService", "router"]

View File

@@ -0,0 +1,35 @@
"""Noms Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import NomsService
from .schemas import NomsResponse
router = APIRouter()
@router.get("/", response_model=List[NomsResponse])
async def search(
fraccion: Optional[str] = Query(None),
pais: Optional[str] = Query(None),
nico: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = NomsService.get_instance()
return await service.search(
fraccion=fraccion, pais=pais, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=NomsResponse)
async def get_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
return await NomsService.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,27 @@
"""NOMs Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class NomsResponse(BaseModel):
"""Normas Oficiales Mexicanas"""
FRACCION: Optional[str] = Field("", max_length=10)
IMPORTACION: Optional[str] = Field("", max_length=1)
EXPORTACION: Optional[str] = Field("", max_length=1)
PERMISO: Optional[str] = Field("", max_length=19)
ACUERDO: Optional[str] = Field("", max_length=100)
CONDICION: Optional[str] = Field("")
FUNDAMENTO: Optional[str] = Field("")
DOF: Optional[str] = Field("", max_length=8)
FORMATONOM: Optional[str] = Field("", max_length=100)
INSTRUCCIONES: Optional[str] = Field("", max_length=100)
CRITERIO: Optional[str] = Field("", max_length=254)
COMPLEMENTO: Optional[str] = Field("", max_length=999)
NICO: Optional[str] = Field("", max_length=2)
PAIS: Optional[str] = Field("", max_length=3)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,37 @@
"""Noms Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import NomsResponse
class NomsService(SitarAPIBaseService):
_instance: Optional["NomsService"] = None
@classmethod
def get_instance(cls) -> "NomsService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
pais: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[NomsResponse]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if pais:
params["pais"] = pais
if nico:
params["nico"] = nico
data = await self._make_request("GET", "/api/v1/noms/", params=params)
return [NomsResponse(**item) for item in data]
async def get_by_id(self, sysid: int) -> NomsResponse:
data = await self._make_request("GET", f"/api/v1/noms/{sysid}")
return NomsResponse(**data)

View File

@@ -0,0 +1,7 @@
"""Precios Estimados Module"""
from .schemas import PreciosEstimados2Response
from .service import PreciosEstimadosService
from .router import router
__all__ = ["PreciosEstimados2Response", "PreciosEstimadosService", "router"]

View File

@@ -0,0 +1,34 @@
"""Precios Estimados Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import PreciosEstimadosService
from .schemas import PreciosEstimados2Response
router = APIRouter()
@router.get("/", response_model=List[PreciosEstimados2Response])
async def search_precios(
fraccion: Optional[str] = Query(None),
nico: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = PreciosEstimadosService.get_instance()
return await service.search(
fraccion=fraccion, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=PreciosEstimados2Response)
async def get_precio_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
return await PreciosEstimadosService.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,18 @@
"""Precios Estimados Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class PreciosEstimados2Response(BaseModel):
"""Precios Estimados"""
FRACCION: Optional[str] = Field(None, max_length=10)
UMC: Optional[str] = Field(None, max_length=2)
PRECIO: Optional[str] = None
DOF: Optional[str] = Field(None, max_length=8)
NICO: Optional[str] = Field("", max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,36 @@
"""Precios Estimados Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import PreciosEstimados2Response
class PreciosEstimadosService(SitarAPIBaseService):
_instance: Optional["PreciosEstimadosService"] = None
@classmethod
def get_instance(cls) -> "PreciosEstimadosService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[PreciosEstimados2Response]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if nico:
params["nico"] = nico
data = await self._make_request(
"GET", "/api/v1/precios-estimados2/", params=params
)
return [PreciosEstimados2Response(**item) for item in data]
async def get_by_id(self, sysid: int) -> PreciosEstimados2Response:
data = await self._make_request("GET", f"/api/v1/precios-estimados2/{sysid}")
return PreciosEstimados2Response(**data)

View File

@@ -0,0 +1,7 @@
"""PROSEC Module"""
from .schemas import ProsecResponse
from .service import ProsecService
from .router import router
__all__ = ["ProsecResponse", "ProsecService", "router"]

View File

@@ -0,0 +1,35 @@
"""PROSEC Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import ProsecService
from .schemas import ProsecResponse
router = APIRouter()
@router.get("/", response_model=List[ProsecResponse])
async def search_prosec(
fraccion: Optional[str] = Query(None),
nico: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = ProsecService.get_instance()
return await service.search(
fraccion=fraccion, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error: {str(e)}")
@router.get("/{sysid}", response_model=ProsecResponse)
async def get_prosec_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
service = ProsecService.get_instance()
return await service.get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=f"Not found: {str(e)}")

View File

@@ -0,0 +1,21 @@
"""PROSEC Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class ProsecResponse(BaseModel):
"""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)
NICO: Optional[str] = Field("", max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,37 @@
"""PROSEC Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import ProsecResponse
class ProsecService(SitarAPIBaseService):
"""Service for PROSEC operations"""
_instance: Optional["ProsecService"] = None
@classmethod
def get_instance(cls) -> "ProsecService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[ProsecResponse]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if nico:
params["nico"] = nico
data = await self._make_request("GET", "/api/v1/prosec/", params=params)
return [ProsecResponse(**item) for item in data]
async def get_by_id(self, sysid: int) -> ProsecResponse:
data = await self._make_request("GET", f"/api/v1/prosec/{sysid}")
return ProsecResponse(**data)

View File

@@ -0,0 +1,7 @@
"""Rcg2 Module"""
from .schemas import Rcg2Response
from .service import Rcg2Service
from .router import router
__all__ = ["Rcg2Response", "Rcg2Service", "router"]

View File

@@ -0,0 +1,34 @@
"""Rcg2 Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import Rcg2Service
from .schemas import Rcg2Response
router = APIRouter()
@router.get("/", response_model=List[Rcg2Response])
async def search(
fraccion: Optional[str] = Query(None),
nico: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = Rcg2Service.get_instance()
return await service.search(
fraccion=fraccion, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=Rcg2Response)
async def get_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
return await Rcg2Service.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,26 @@
"""RCG2 Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class Rcg2Response(BaseModel):
"""Reglas de Carácter General"""
FRACCION: Optional[str] = Field(None, max_length=10)
ANEXO: Optional[str] = Field(None, max_length=19)
SECTOR: Optional[str] = None
REGIMEN: Optional[str] = None
CONDICION: Optional[str] = None
ADUANAS: Optional[str] = Field(None, max_length=1000)
DOF: Optional[str] = Field(None, max_length=8)
ARCHIVOCRITERIO: Optional[str] = Field(None, max_length=1000)
TIPO: Optional[str] = Field(None, max_length=1)
IMPO: Optional[int] = None
EXPO: Optional[int] = None
WEB_DOCUMENTO_ID: Optional[int] = None
NICO: Optional[str] = Field("", max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,34 @@
"""Rcg2 Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import Rcg2Response
class Rcg2Service(SitarAPIBaseService):
_instance: Optional["Rcg2Service"] = None
@classmethod
def get_instance(cls) -> "Rcg2Service":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[Rcg2Response]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if nico:
params["nico"] = nico
data = await self._make_request("GET", "/api/v1/rcg2/", params=params)
return [Rcg2Response(**item) for item in data]
async def get_by_id(self, sysid: int) -> Rcg2Response:
data = await self._make_request("GET", f"/api/v1/rcg2/{sysid}")
return Rcg2Response(**data)

View File

@@ -0,0 +1,7 @@
"""Regulaciones Module"""
from .schemas import RegulacionesResponse
from .service import RegulacionesService
from .router import router
__all__ = ["RegulacionesResponse", "RegulacionesService", "router"]

View File

@@ -0,0 +1,37 @@
"""Regulaciones Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import RegulacionesService
from .schemas import RegulacionesResponse
router = APIRouter()
@router.get("/", response_model=List[RegulacionesResponse])
async def search_regulaciones(
fraccion: Optional[str] = Query(None, description="Fracción arancelaria"),
nico: Optional[str] = Query(None, description="NICO"),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = RegulacionesService.get_instance()
return await service.search(
fraccion=fraccion, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error: {str(e)}")
@router.get("/{sysid}", response_model=RegulacionesResponse)
async def get_regulacion_by_id(
sysid: int, current_user: dict = Depends(get_current_user)
):
try:
service = RegulacionesService.get_instance()
return await service.get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=f"Not found: {str(e)}")

View File

@@ -0,0 +1,19 @@
"""Regulaciones Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class RegulacionesResponse(BaseModel):
"""Regulaciones y restricciones"""
FRACCION: Optional[str] = Field(None, max_length=10)
PERMISO: Optional[str] = Field(None, max_length=2)
ACUERDO: Optional[str] = Field(None, max_length=15)
CLAVE: Optional[str] = Field(None, max_length=10)
DESCRIPCION: Optional[str] = None
NICO: Optional[str] = Field("", max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,37 @@
"""Regulaciones Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import RegulacionesResponse
class RegulacionesService(SitarAPIBaseService):
"""Service for Regulaciones operations"""
_instance: Optional["RegulacionesService"] = None
@classmethod
def get_instance(cls) -> "RegulacionesService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[RegulacionesResponse]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if nico:
params["nico"] = nico
data = await self._make_request("GET", "/api/v1/regulaciones/", params=params)
return [RegulacionesResponse(**item) for item in data]
async def get_by_id(self, sysid: int) -> RegulacionesResponse:
data = await self._make_request("GET", f"/api/v1/regulaciones/{sysid}")
return RegulacionesResponse(**data)

View File

@@ -0,0 +1,7 @@
"""Reit Module"""
from .schemas import ReitResponse
from .service import ReitService
from .router import router
__all__ = ["ReitResponse", "ReitService", "router"]

View File

@@ -0,0 +1,34 @@
"""Reit Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import ReitService
from .schemas import ReitResponse
router = APIRouter()
@router.get("/", response_model=List[ReitResponse])
async def search(
fraccion: Optional[str] = Query(None),
nico: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = ReitService.get_instance()
return await service.search(
fraccion=fraccion, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=ReitResponse)
async def get_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
return await ReitService.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,25 @@
"""REIT Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class ReitResponse(BaseModel):
"""Registro de Empresas de Industria Terminal"""
FRACCION: Optional[str] = Field(None, max_length=10)
ARTICULO: Optional[str] = Field(None, max_length=50)
FUNDAMENTO: Optional[str] = Field(None, max_length=1500)
ACUERDO: Optional[str] = Field(None, max_length=500)
PERMISO: Optional[str] = Field(None, max_length=2)
DOF: Optional[str] = Field(None, max_length=8)
DOCUMENTO: Optional[str] = Field(None, max_length=200)
TEMPORALIDAD: Optional[int] = None
TEMPORALIDADSERVICIO: Optional[int] = None
TEMPORALIDADCERTIFICADA: Optional[int] = None
WEB_DOCUMENTO_ID: Optional[int] = None
NICO: Optional[str] = Field("", max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,34 @@
"""Reit Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import ReitResponse
class ReitService(SitarAPIBaseService):
_instance: Optional["ReitService"] = None
@classmethod
def get_instance(cls) -> "ReitService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[ReitResponse]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if nico:
params["nico"] = nico
data = await self._make_request("GET", "/api/v1/reit/", params=params)
return [ReitResponse(**item) for item in data]
async def get_by_id(self, sysid: int) -> ReitResponse:
data = await self._make_request("GET", f"/api/v1/reit/{sysid}")
return ReitResponse(**data)

View File

@@ -0,0 +1,7 @@
"""RequisitoPrevio Module"""
from .schemas import RequisitoPrevioResponse
from .service import RequisitoPrevioService
from .router import router
__all__ = ["RequisitoPrevioResponse", "RequisitoPrevioService", "router"]

View File

@@ -0,0 +1,34 @@
"""RequisitoPrevio Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import RequisitoPrevioService
from .schemas import RequisitoPrevioResponse
router = APIRouter()
@router.get("/", response_model=List[RequisitoPrevioResponse])
async def search(
fraccion: Optional[str] = Query(None),
nico: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = RequisitoPrevioService.get_instance()
return await service.search(
fraccion=fraccion, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=RequisitoPrevioResponse)
async def get_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
return await RequisitoPrevioService.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,28 @@
"""Requisito Previo Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class RequisitoPrevioResponse(BaseModel):
"""Requisitos previos"""
FRACCION: Optional[str] = Field(None, max_length=8)
DESCRIPCION: Optional[str] = Field(None, max_length=1999)
OBSERVACION: Optional[str] = Field(None, max_length=1999)
DOF: Optional[str] = Field(None, max_length=8)
VIGENCIA: Optional[str] = Field(None, max_length=8)
NUMEROACUERDO: Optional[int] = None
NUMEROCRITERIO: Optional[int] = None
NUMEROFORMATO: Optional[int] = None
NUMEROINSTRUCCIONES: Optional[int] = None
HISTORICO: Optional[str] = Field(None, max_length=2000)
NOTA: Optional[str] = Field(None, max_length=5000)
PERMISO: Optional[str] = Field(None, max_length=2)
IMPO: Optional[int] = None
EXPO: Optional[int] = None
NICO: Optional[str] = Field("", max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,36 @@
"""RequisitoPrevio Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import RequisitoPrevioResponse
class RequisitoPrevioService(SitarAPIBaseService):
_instance: Optional["RequisitoPrevioService"] = None
@classmethod
def get_instance(cls) -> "RequisitoPrevioService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[RequisitoPrevioResponse]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if nico:
params["nico"] = nico
data = await self._make_request(
"GET", "/api/v1/requisito-previo/", params=params
)
return [RequisitoPrevioResponse(**item) for item in data]
async def get_by_id(self, sysid: int) -> RequisitoPrevioResponse:
data = await self._make_request("GET", f"/api/v1/requisito-previo/{sysid}")
return RequisitoPrevioResponse(**data)

View File

@@ -0,0 +1,90 @@
"""
SITAR API Main Router
Router principal que incluye todos los sub-routers de SITAR.
Registra este router en tu aplicación FastAPI principal.
Usage:
from api.v1.modules.sitar.main_router import router as sitar_router
app.include_router(
sitar_router,
prefix="/api/v1/sitar",
tags=["sitar"]
)
"""
from fastapi import APIRouter
# Import all sub-routers
from .tlcs.router import router as tlcs_router
from .fracciones.router import router as fracciones_router
from .fracciones_usa.router import router as fracciones_usa_router
from .regulaciones.router import router as regulaciones_router
from .prosec.router import router as prosec_router
from .precios_estimados.router import router as precios_estimados_router
from .fundamentos_tlc.router import router as fundamentos_tlc_router
from .aladi2.router import router as aladi2_router
from .cuotas2.router import router as cuotas2_router
from .cupos.router import router as cupos_router
from .fracciones_anteriores.router import router as fracciones_anteriores_router
from .informacion_general.router import router as informacion_general_router
from .ieps.router import router as ieps_router
from .noms.router import router as noms_router
from .rcg2.router import router as rcg2_router
from .reit.router import router as reit_router
from .requisito_previo.router import router as requisito_previo_router
from .vehiculos_marcas.router import router as vehiculos_marcas_router
from .vehiculos_modelos.router import router as vehiculos_modelos_router
# Create main router
router = APIRouter()
# Include all sub-routers with their prefixes
router.include_router(tlcs_router, prefix="/tlcs", tags=["sitar-tlcs"])
router.include_router(
fracciones_router, prefix="/fracciones", tags=["sitar-fracciones"]
)
router.include_router(
fracciones_usa_router, prefix="/fracciones-usa", tags=["sitar-fracciones-usa"]
)
router.include_router(
regulaciones_router, prefix="/regulaciones", tags=["sitar-regulaciones"]
)
router.include_router(prosec_router, prefix="/prosec", tags=["sitar-prosec"])
router.include_router(
precios_estimados_router,
prefix="/precios-estimados",
tags=["sitar-precios-estimados"],
)
router.include_router(
fundamentos_tlc_router, prefix="/fundamentos-tlc", tags=["sitar-fundamentos-tlc"]
)
router.include_router(aladi2_router, prefix="/aladi2", tags=["sitar-aladi2"])
router.include_router(cuotas2_router, prefix="/cuotas2", tags=["sitar-cuotas2"])
router.include_router(cupos_router, prefix="/cupos", tags=["sitar-cupos"])
router.include_router(
fracciones_anteriores_router,
prefix="/fracciones-anteriores",
tags=["sitar-fracciones-anteriores"],
)
router.include_router(
informacion_general_router,
prefix="/informacion-general",
tags=["sitar-informacion-general"],
)
router.include_router(ieps_router, prefix="/ieps", tags=["sitar-ieps"])
router.include_router(noms_router, prefix="/noms", tags=["sitar-noms"])
router.include_router(rcg2_router, prefix="/rcg2", tags=["sitar-rcg2"])
router.include_router(reit_router, prefix="/reit", tags=["sitar-reit"])
router.include_router(
requisito_previo_router, prefix="/requisito-previo", tags=["sitar-requisito-previo"]
)
router.include_router(
vehiculos_marcas_router, prefix="/vehiculos-marcas", tags=["sitar-vehiculos-marcas"]
)
router.include_router(
vehiculos_modelos_router,
prefix="/vehiculos-modelos",
tags=["sitar-vehiculos-modelos"],
)

View File

@@ -0,0 +1,7 @@
"""TLCS Module"""
from .schemas import TLCSResponse
from .service import TLCSService
from .router import router
__all__ = ["TLCSResponse", "TLCSService", "router"]

View File

@@ -0,0 +1,44 @@
"""TLCS Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import TLCSService
from .schemas import TLCSResponse
router = APIRouter()
@router.get("/", response_model=List[TLCSResponse])
async def search_tlcs(
fraccion: Optional[str] = Query(None, description="Fracción arancelaria"),
pais: Optional[str] = Query(None, description="Código de país"),
nico: Optional[str] = Query(None, description="NICO"),
skip: int = Query(0, ge=0, description="Registros a saltar"),
limit: int = Query(100, ge=1, le=1000, description="Máximo de registros"),
current_user: dict = Depends(get_current_user),
):
"""Search TLCS records"""
try:
service = TLCSService.get_instance()
return await service.search(
fraccion=fraccion, pais=pais, nico=nico, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Error fetching TLCS data: {str(e)}"
)
@router.get("/{sysid}/{fraccion}", response_model=TLCSResponse)
async def get_tlcs_by_id(
sysid: int,
fraccion: str,
current_user: dict = Depends(get_current_user),
):
"""Get single TLCS record by SYSID and FRACCION"""
try:
service = TLCSService.get_instance()
return await service.get_by_id(sysid, fraccion)
except Exception as e:
raise HTTPException(status_code=404, detail=f"TLCS record not found: {str(e)}")

View File

@@ -0,0 +1,23 @@
"""TLCS Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class TLCSResponse(BaseModel):
"""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
DOF: Optional[str] = Field(None, max_length=8)
OBSERVACION: Optional[str] = None
NICO: Optional[str] = Field("", max_length=2)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,55 @@
"""TLCS Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import TLCSResponse
class TLCSService(SitarAPIBaseService):
"""Service for TLCS operations"""
_instance: Optional["TLCSService"] = None
@classmethod
def get_instance(cls) -> "TLCSService":
"""Get singleton instance"""
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
pais: Optional[str] = None,
nico: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[TLCSResponse]:
"""
Search TLCS records
Args:
fraccion: Tariff fraction code
pais: Country code
nico: NICO code
skip: Records to skip
limit: Maximum records to return
Returns:
List of TLCS records
"""
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion[:8]
if pais:
params["pais"] = pais
if nico:
params["nico"] = nico
data = await self._make_request("GET", "/api/v1/tlcs/", params=params)
return [TLCSResponse(**item) for item in data]
async def get_by_id(self, sysid: int, fraccion: str) -> TLCSResponse:
"""Get single TLCS record by SYSID and FRACCION"""
data = await self._make_request("GET", f"/api/v1/tlcs/{sysid}/{fraccion}")
return TLCSResponse(**data)

View File

@@ -0,0 +1,7 @@
"""VehiculosMarcas Module"""
from .schemas import VehiculosMarcasResponse
from .service import VehiculosMarcasService
from .router import router
__all__ = ["VehiculosMarcasResponse", "VehiculosMarcasService", "router"]

View File

@@ -0,0 +1,34 @@
"""VehiculosMarcas Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import VehiculosMarcasService
from .schemas import VehiculosMarcasResponse
router = APIRouter()
@router.get("/", response_model=List[VehiculosMarcasResponse])
async def search(
fraccion: Optional[str] = Query(None),
marca: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = VehiculosMarcasService.get_instance()
return await service.search(
fraccion=fraccion, marca=marca, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=VehiculosMarcasResponse)
async def get_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
return await VehiculosMarcasService.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,15 @@
"""Vehiculos Marcas Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class VehiculosMarcasResponse(BaseModel):
"""Marcas de vehículos"""
FRACCION: Optional[str] = Field(None, max_length=8)
MARCA: Optional[str] = Field(None, max_length=60)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,36 @@
"""VehiculosMarcas Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import VehiculosMarcasResponse
class VehiculosMarcasService(SitarAPIBaseService):
_instance: Optional["VehiculosMarcasService"] = None
@classmethod
def get_instance(cls) -> "VehiculosMarcasService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
marca: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[VehiculosMarcasResponse]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if marca:
params["marca"] = marca
data = await self._make_request(
"GET", "/api/v1/vehiculos-marcas/", params=params
)
return [VehiculosMarcasResponse(**item) for item in data]
async def get_by_id(self, sysid: int) -> VehiculosMarcasResponse:
data = await self._make_request("GET", f"/api/v1/vehiculos-marcas/{sysid}")
return VehiculosMarcasResponse(**data)

View File

@@ -0,0 +1,7 @@
"""VehiculosModelos Module"""
from .schemas import VehiculosModelosResponse
from .service import VehiculosModelosService
from .router import router
__all__ = ["VehiculosModelosResponse", "VehiculosModelosService", "router"]

View File

@@ -0,0 +1,35 @@
"""VehiculosModelos Router"""
from typing import Optional, List
from fastapi import APIRouter, HTTPException, Query, Depends
from core.security import get_current_user
from .service import VehiculosModelosService
from .schemas import VehiculosModelosResponse
router = APIRouter()
@router.get("/", response_model=List[VehiculosModelosResponse])
async def search(
fraccion: Optional[str] = Query(None),
marca: Optional[str] = Query(None),
modelo: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
current_user: dict = Depends(get_current_user),
):
try:
service = VehiculosModelosService.get_instance()
return await service.search(
fraccion=fraccion, marca=marca, modelo=modelo, skip=skip, limit=limit
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{sysid}", response_model=VehiculosModelosResponse)
async def get_by_id(sysid: int, current_user: dict = Depends(get_current_user)):
try:
return await VehiculosModelosService.get_instance().get_by_id(sysid)
except Exception as e:
raise HTTPException(status_code=404, detail=str(e))

View File

@@ -0,0 +1,38 @@
"""Vehiculos Modelos Schemas"""
from typing import Optional
from pydantic import BaseModel, Field
class VehiculosModelosResponse(BaseModel):
"""Modelos de vehículos"""
FRACCION: Optional[str] = Field(None, max_length=8)
MARCA: Optional[str] = Field(None, max_length=60)
MODELO: Optional[str] = Field(None, max_length=80)
TIPO: Optional[str] = Field(None, max_length=1)
CLAVEUM: Optional[str] = Field(None, max_length=2)
A20: Optional[str] = Field(None, max_length=19)
A19: Optional[str] = Field(None, max_length=19)
A18: Optional[str] = Field(None, max_length=19)
A17: Optional[str] = Field(None, max_length=19)
A16: Optional[str] = Field(None, max_length=19)
A15: Optional[str] = Field(None, max_length=19)
A14: Optional[str] = Field(None, max_length=19)
A13: Optional[str] = Field(None, max_length=19)
A12: Optional[str] = Field(None, max_length=19)
A11: Optional[str] = Field(None, max_length=19)
A10: Optional[str] = Field(None, max_length=19)
A9: Optional[str] = Field(None, max_length=19)
A8: Optional[str] = Field(None, max_length=19)
A7: Optional[str] = Field(None, max_length=19)
A6: Optional[str] = Field(None, max_length=19)
A5: Optional[str] = Field(None, max_length=19)
A4: Optional[str] = Field(None, max_length=19)
A3: Optional[str] = Field(None, max_length=19)
A2: Optional[str] = Field(None, max_length=19)
A1: Optional[str] = Field(None, max_length=19)
SYSID: int
class Config:
from_attributes = True

View File

@@ -0,0 +1,39 @@
"""VehiculosModelos Service"""
from typing import Optional, List
from ..common import SitarAPIBaseService
from .schemas import VehiculosModelosResponse
class VehiculosModelosService(SitarAPIBaseService):
_instance: Optional["VehiculosModelosService"] = None
@classmethod
def get_instance(cls) -> "VehiculosModelosService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
async def search(
self,
fraccion: Optional[str] = None,
marca: Optional[str] = None,
modelo: Optional[str] = None,
skip: int = 0,
limit: int = 100,
) -> List[VehiculosModelosResponse]:
params = {"skip": skip, "limit": min(limit, 1000)}
if fraccion:
params["fraccion"] = fraccion
if marca:
params["marca"] = marca
if modelo:
params["modelo"] = modelo
data = await self._make_request(
"GET", "/api/v1/vehiculos-modelos/", params=params
)
return [VehiculosModelosResponse(**item) for item in data]
async def get_by_id(self, sysid: int) -> VehiculosModelosResponse:
data = await self._make_request("GET", f"/api/v1/vehiculos-modelos/{sysid}")
return VehiculosModelosResponse(**data)