78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""Fracciones USA Service"""
|
|
|
|
import logging
|
|
from typing import Optional, List
|
|
|
|
from ..common import SitarAPIBaseService
|
|
from .schemas import FraccionesUSAResponse
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
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,
|
|
descripcion: 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
|
|
if descripcion:
|
|
params["descripcion"] = descripcion
|
|
|
|
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)
|
|
|
|
@classmethod
|
|
def search_sync(
|
|
cls,
|
|
fraccion: Optional[str] = None,
|
|
descripcion: Optional[str] = None,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> List[FraccionesUSAResponse]:
|
|
"""
|
|
Synchronous SITAR search for use from sync validators (FastAPI async routes
|
|
run sync code on the event loop; asyncio.run must not be used there).
|
|
"""
|
|
try:
|
|
service = cls.get_instance()
|
|
except ValueError:
|
|
return []
|
|
|
|
params: dict = {"skip": skip, "limit": min(limit, 1000)}
|
|
if fraccion:
|
|
params["fraccion"] = fraccion
|
|
if descripcion:
|
|
params["descripcion"] = descripcion
|
|
|
|
try:
|
|
data = service._make_request_sync("GET", "api/v1/fracciones-usa/", params=params)
|
|
if not isinstance(data, list):
|
|
return []
|
|
return [FraccionesUSAResponse(**item) for item in data]
|
|
except Exception as exc:
|
|
logger.warning("SITAR fracciones-usa search_sync failed: %s", exc)
|
|
return []
|