a4cf394447
- Changed the default environment variable from 'development' to 'production' in docker-compose.prod.yml and docker-compose.yml. - Added SITAR API credentials to the environment variables in both Docker Compose files. - Updated the seed data for customs units of measure to include a76_unit_code. - Refactored the UnitOfMeasureCustoms model and related DTOs to replace scaii_unit_code with a76_unit_code for consistency. - Adjusted frontend components to reflect the updated unit of measure structure and ensure proper handling of the new a76_unit_code field.
71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
"""Fracciones Service"""
|
|
|
|
import asyncio
|
|
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,
|
|
description: Optional[str] = None,
|
|
nivel: Optional[int] = 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
|
|
if description:
|
|
params["descripcion"] = description
|
|
if nivel is not None:
|
|
params["nivel"] = nivel
|
|
|
|
data = await self._make_request("GET", "/api/v1/fracciones/", params=params)
|
|
return [FraccionesResponse(**item) for item in data]
|
|
|
|
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)
|
|
|
|
@classmethod
|
|
def search_sync(
|
|
cls,
|
|
fraccion: Optional[str] = None,
|
|
nico: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
nivel: Optional[int] = None,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> List[FraccionesResponse]:
|
|
"""Search Mexican tariff fractions (sync wrapper for use in Celery/sync context)."""
|
|
service = cls.get_instance()
|
|
return asyncio.run(
|
|
service.search(
|
|
fraccion=fraccion,
|
|
nico=nico,
|
|
description=description,
|
|
nivel=nivel,
|
|
skip=skip,
|
|
limit=limit,
|
|
)
|
|
)
|