ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'development' of https://git.aduanasoft.com/ADUANASOFT/anexo76 into feature/manuales

Browse Source
This commit is contained in:
Kevin_Ramirez
2026-04-29 10:25:29 -05:00
parent 97de088b5d a4c9c655a6
commit 057eaa50e6
118 changed files with 8389 additions and 5951 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
backend/alembic/versions/b2c3d4e5f6a7_drop_client_id_from_parts.py Normal file
View File

@@ -0,0 +1,28 @@
"""drop client_id column from parts
Revision ID: b2c3d4e5f6a7
Revises: a1b2c3d4e5f6
Create Date: 2026-04-28 11:50:00.000000
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = "b2c3d4e5f6a7"
down_revision = "a1b2c3d4e5f6"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.drop_column("parts", "client_id", schema="a76")
def downgrade() -> None:
op.add_column(
"parts",
sa.Column("client_id", sa.Integer(), nullable=True),
schema="a76",
)

29
backend/alembic/versions/c3d4e5f6a7b8_add_action_to_doda_alta_log.py Normal file
View File

@@ -0,0 +1,29 @@
"""add action column to doda_alta_log
Revision ID: c3d4e5f6a7b8
Revises: b2c3d4e5f6a7
Create Date: 2026-04-28 13:20:00.000000
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = "c3d4e5f6a7b8"
down_revision = "b2c3d4e5f6a7"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column(
"doda_alta_log",
sa.Column("action", sa.String(length=20), nullable=True),
schema="a76",
)
def downgrade() -> None:
op.drop_column("doda_alta_log", "action", schema="a76")

8
backend/api/v1/modules/a76/expediente_archivos/routes.py
View File

@@ -60,11 +60,17 @@ def list_expediente_archivos(
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=200),
search: str = Query(None),
status: str = Query(None),
rfc_consulta: str = Query(None),
e_document: str = Query(None),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
tenant_id = validate_access_to_resource(db, company_id, current_user)
return ExpedienteArchivoService.list(db, company_id, tenant_id, page, page_size, search)
return ExpedienteArchivoService.list(
db, company_id, tenant_id, page, page_size,
search=search, status=status, rfc_consulta=rfc_consulta, e_document=e_document
)
@router.get("/{record_id}", response_model=ExpedienteArchivoResponseDTO)

18
backend/api/v1/modules/a76/expediente_archivos/service.py
View File

@@ -75,6 +75,9 @@ class ExpedienteArchivoService:
page: int = 1,
page_size: int = 50,
search: Optional[str] = None,
status: Optional[str] = None,
rfc_consulta: Optional[str] = None,
e_document: Optional[str] = None,
) -> ExpedienteArchivoListResponse:
query = (
db.query(ExpedienteArchivo)
@@ -84,11 +87,22 @@ class ExpedienteArchivoService:
ExpedienteArchivo.deleted_at.is_(None),
)
)
if status:
query = query.filter(ExpedienteArchivo.status == status)
if rfc_consulta:
query = query.filter(ExpedienteArchivo.rfc_consulta.ilike(f"%{rfc_consulta}%"))
if e_document:
query = query.filter(ExpedienteArchivo.e_document.ilike(f"%{e_document}%"))
if search:
like = f"%{search}%"
query = query.filter(
ExpedienteArchivo.e_document.ilike(like)
| ExpedienteArchivo.tipo_documento.ilike(like)
or_(
ExpedienteArchivo.e_document.ilike(like),
ExpedienteArchivo.tipo_documento.ilike(like),
ExpedienteArchivo.rfc_consulta.ilike(like),
ExpedienteArchivo.num_operacion.ilike(like),
ExpedienteArchivo.nombre_archivo.ilike(like),
)
)
total = query.count()
items = query.order_by(ExpedienteArchivo.id.desc()).offset((page - 1) * page_size).limit(page_size).all()

1
backend/api/v1/modules/a76/general_catalogs/concepts/routes.py
View File

@@ -11,6 +11,7 @@ router = TenantCRUDRoutes(
tags=["a76.general_catalogs.concepts"],
resource_name="Concept",
enable_list=True,
enable_filters=True,
list_permissions=["cat_concepts.view"],
get_permissions=["cat_concepts.view"],
create_permissions=["cat_concepts.create"],

2
backend/api/v1/modules/a76/general_catalogs/doda/alta_log_dto.py
View File

@@ -8,6 +8,7 @@ from pydantic import BaseModel, Field
class DodaAltaLogCreateDTO(BaseModel):
doda_id: Optional[int] = None
variant: Optional[str] = Field(None, max_length=10)
action: Optional[str] = Field(None, max_length=20)
responsible: Optional[str] = Field(None, max_length=20)
patent: Optional[str] = Field(None, max_length=10)
dispatch_customs: Optional[str] = Field(None, max_length=10)
@@ -29,6 +30,7 @@ class DodaAltaLogResponseDTO(BaseModel):
id: int
doda_id: Optional[int] = None
variant: Optional[str] = None
action: Optional[str] = None
responsible: Optional[str] = None
patent: Optional[str] = None
dispatch_customs: Optional[str] = None

1
backend/api/v1/modules/a76/general_catalogs/doda/alta_log_models.py
View File

@@ -23,6 +23,7 @@ class DodaAltaLog(Base, TenantScopedMixin, TimestampMixin):
# Tipo de alta (doda / pita)
variant: Mapped[str | None] = mapped_column(String(10), nullable=True)
action: Mapped[str | None] = mapped_column(String(20), nullable=True)
# Datos copiados del DODA al momento del envío (para historial)
responsible: Mapped[str | None] = mapped_column(String(20), nullable=True)

2
backend/api/v1/modules/a76/general_catalogs/doda/alta_log_service.py
View File

@@ -115,6 +115,7 @@ class DodaAltaLogService:
tenant_id: int,
variant: str,
ext_result: dict,
action: str = "alta",
) -> DodaAltaLog:
"""
Crea un registro de log a partir de la respuesta del servicio externo de alta.
@@ -127,6 +128,7 @@ class DodaAltaLogService:
dto = DodaAltaLogCreateDTO(
doda_id=doda.id,
variant=variant,
action=action,
responsible=doda.responsible,
patent=doda.patent,
dispatch_customs=doda.dispatch_customs,

92
backend/api/v1/modules/a76/general_catalogs/doda/alta_service.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import base64
import json
import logging
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
@@ -16,6 +17,7 @@ from core.storage_s3 import get_object_bytes, object_exists
from api.v1.modules.a76.customs_brokers import models as cb_models
from .models import Doda, DodaContainer, DodaAmericanPedimento, DodaPedimento
from .alta_log_models import DodaAltaLog
from .payload_normalizer import (
normalize_aduana_despacho,
normalize_aduana_seccion,
@@ -487,6 +489,40 @@ class DodaAltaService:
# Build full payload
# ------------------------------------------------------------------
def _latest_alta_log(
self,
doda_id: int,
tenant_id: int,
company_id: int,
variant: str,
) -> Optional[DodaAltaLog]:
return (
self.db.query(DodaAltaLog)
.filter(
DodaAltaLog.doda_id == doda_id,
DodaAltaLog.tenant_id == tenant_id,
DodaAltaLog.company_id == company_id,
DodaAltaLog.variant == variant,
DodaAltaLog.deleted_at.is_(None),
)
.order_by(DodaAltaLog.id.desc())
.first()
)
@staticmethod
def _extract_numero_transaccion(log_record: DodaAltaLog) -> str:
raw_json = (log_record.result_json or "").strip()
if raw_json:
try:
parsed = json.loads(raw_json)
for key in ("numero_transaccion", "transaction_number"):
value = parsed.get(key)
if value:
return str(value).strip()
except Exception:
logger.warning("No se pudo parsear result_json de DodaAltaLog id=%s", log_record.id)
return ""
def build_alta_payload(
self,
doda_id: int,
@@ -549,3 +585,59 @@ class DodaAltaService:
}
return payload
def build_consulta_payload(
self,
doda_id: int,
tenant_id: int,
company_id: int,
variant: str = "doda",
user_email: str = "",
) -> Dict[str, Any]:
payload = self.build_alta_payload(
doda_id=doda_id,
tenant_id=tenant_id,
company_id=company_id,
variant=variant,
user_email=user_email,
)
latest_log = self._latest_alta_log(doda_id, tenant_id, company_id, variant)
if not latest_log:
raise ValueError(
"No existe un alta DODA previa para construir la consulta (falta task/log)."
)
numero_transaccion = self._extract_numero_transaccion(latest_log)
if not numero_transaccion:
raise ValueError(
"No se encontro numero_transaccion en el ultimo resultado de alta DODA."
)
payload["numero_transaccion"] = numero_transaccion
return payload
def build_eliminar_payload(
self,
doda_id: int,
tenant_id: int,
company_id: int,
variant: str = "doda",
user_email: str = "",
) -> Dict[str, Any]:
payload = self.build_alta_payload(
doda_id=doda_id,
tenant_id=tenant_id,
company_id=company_id,
variant=variant,
user_email=user_email,
)
latest_log = self._latest_alta_log(doda_id, tenant_id, company_id, variant)
if not latest_log:
raise ValueError(
"No existe un alta DODA previa para construir la eliminacion."
)
numero_integracion = (latest_log.integration_number or "").strip()
if not numero_integracion:
raise ValueError(
"No se encontro numero_integracion en el historial de alta DODA."
)
payload["numero_integracion"] = numero_integracion
return payload

44
backend/api/v1/modules/a76/general_catalogs/doda/external_service.py
View File

@@ -17,6 +17,10 @@ class DodaExternalService:
Endpoints:
POST {base_url}/api/v1/doda/alta
GET {base_url}/api/v1/doda/alta-status/{task_id}
POST {base_url}/api/v1/doda/consulta
GET {base_url}/api/v1/doda/consulta-status/{task_id}
POST {base_url}/api/v1/doda/eliminar
GET {base_url}/api/v1/doda/eliminar-status/{task_id}
Usa COVE_API_URL como URL base (la misma variable que COVE y Expediente).
"""
@@ -61,3 +65,43 @@ class DodaExternalService:
response = client.get(url)
response.raise_for_status()
return response.json()
def post_consulta(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Envia consulta DODA y retorna {task_id, status, message}."""
url = f"{self.base_url.rstrip('/')}/api/v1/doda/consulta"
with httpx.Client(
timeout=httpx.Timeout(60.0, connect=10.0), verify=self.verify_ssl
) as client:
response = client.post(url, json=payload)
response.raise_for_status()
return response.json()
def get_consulta_status(self, task_id: str) -> Dict[str, Any]:
"""Consulta el estado de una tarea de consulta DODA."""
url = f"{self.base_url.rstrip('/')}/api/v1/doda/consulta-status/{task_id}"
with httpx.Client(
timeout=httpx.Timeout(30.0, connect=10.0), verify=self.verify_ssl
) as client:
response = client.get(url)
response.raise_for_status()
return response.json()
def post_eliminar(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Envia eliminacion DODA y retorna {task_id, status, message}."""
url = f"{self.base_url.rstrip('/')}/api/v1/doda/eliminar"
with httpx.Client(
timeout=httpx.Timeout(60.0, connect=10.0), verify=self.verify_ssl
) as client:
response = client.post(url, json=payload)
response.raise_for_status()
return response.json()
def get_eliminar_status(self, task_id: str) -> Dict[str, Any]:
"""Consulta el estado de una tarea de eliminacion DODA."""
url = f"{self.base_url.rstrip('/')}/api/v1/doda/eliminar-status/{task_id}"
with httpx.Client(
timeout=httpx.Timeout(30.0, connect=10.0), verify=self.verify_ssl
) as client:
response = client.get(url)
response.raise_for_status()
return response.json()

316
backend/api/v1/modules/a76/general_catalogs/doda/routes.py
View File

@@ -60,6 +60,47 @@ from core.security import get_current_user, validate_access_to_resource
logger = logging.getLogger(__name__)
def _coalesce_external_result_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
result = payload.get("result")
if isinstance(result, dict):
return result
return payload
def _apply_consulta_success_to_doda(
doda: Doda,
payload: Dict[str, Any],
) -> None:
source = _coalesce_external_result_payload(payload)
mapping = {
"integration_number": "integration_number",
"numero_integracion": "integration_number",
"transaction_number": "transaction_number",
"numero_transaccion": "transaction_number",
"sat_digital_seal": "sat_digital_seal",
"sello_digital_sat": "sat_digital_seal",
"sat_certificate": "sat_certificate",
"certificado_sat": "sat_certificate",
"serial_number": "serial_number",
"numero_serie": "serial_number",
"electronic_signature": "electronic_signature",
"firma_electronica": "electronic_signature",
"original_chain": "original_chain",
"cadena_original": "original_chain",
"sat_original_chain": "sat_original_chain",
"cadena_original_sat": "sat_original_chain",
"linq_sat_qr": "linq_sat_qr",
"link_sat_qr": "linq_sat_qr",
"xml_doda_sent_path": "xml_doda_sent_path",
"xml_doda_response_path": "xml_doda_response_path",
"status": "status",
}
for src_key, dst_attr in mapping.items():
value = source.get(src_key)
if value is not None and value != "":
setattr(doda, dst_attr, str(value))
# Router independiente para rutas literales (deben registrarse antes que /{id})
router = APIRouter(prefix="/doda", tags=["doda"])
@@ -82,7 +123,7 @@ async def export_doda_list(
"""
Listado al estilo legacy: filtra `doda_date` (YYYYMMDD) entre inicio y fin.
"""
tenant_id = int(validate_access_to_resource(db, company_id, current_user))
tenant_id = int(validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"]))
try:
d0, d1, fmt, mode = parse_export_params(
date_from, date_to, file_format, date_mode
@@ -126,7 +167,7 @@ async def export_doda_pedimentos(
Reporte por DODA seleccionado: columnas alineadas al listado de pedimentos (PATENTE, DOCUMENTO, COVE, etc.).
Si no hay líneas, se devuelve el archivo solo con encabezados.
"""
tenant_id = int(validate_access_to_resource(db, company_id, current_user))
tenant_id = int(validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"]))
try:
fmt = parse_pedimento_export_format(file_format)
except ValueError as e:
@@ -168,6 +209,11 @@ _crud_router = TenantCRUDRoutes(
id_name="doda_id",
enable_list=True,
enable_filters=True,
list_permissions=["cat_doda.view"],
get_permissions=["cat_doda.view"],
create_permissions=["cat_doda.create"],
update_permissions=["cat_doda.edit"],
delete_permissions=["cat_doda.delete"],
).router
router.include_router(_crud_router)
@@ -183,7 +229,7 @@ async def get_doda_detail(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
tenant_id = validate_access_to_resource(db, company_id, current_user)
tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"])
doda = DodaService.get_by_id(db, doda_id, tenant_id, company_id)
if not doda:
raise HTTPException(
@@ -633,6 +679,7 @@ async def post_doda_alta(
tenant_id=int(tenant_id),
variant=variant,
ext_result=result,
action="alta",
)
except Exception:
logger.exception("Error persistiendo DodaAltaLog para doda_id=%s", doda_id)
@@ -668,6 +715,269 @@ async def get_doda_alta_status(
) from exc
@router.post(
"/{doda_id}/consulta",
summary="Enviar Consulta DODA al servicio externo (asíncrono)",
tags=["doda-alta"],
)
async def post_doda_consulta(
doda_id: int,
company_id: int = Query(..., description="Company ID"),
variant: str = Query("doda", description="Tipo de alta: 'doda' o 'pita'"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
) -> Dict[str, Any]:
tenant_id = validate_access_to_resource(db, company_id, current_user)
user_email = (
current_user.get("email")
or current_user.get("preferred_username")
or ""
)
service = DodaAltaService(db)
try:
payload = service.build_consulta_payload(
doda_id=doda_id,
tenant_id=int(tenant_id),
company_id=company_id,
variant=variant,
user_email=user_email,
)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=str(exc),
) from exc
ext_result: Dict[str, Any]
try:
ext = DodaExternalService()
ext_result = ext.post_consulta(payload)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=str(exc),
) from exc
except Exception as exc:
logger.exception("Error al enviar consulta DODA al servicio externo: doda_id=%s", doda_id)
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"Error al contactar el servicio DODA externo (consulta): {exc}",
) from exc
doda_record = DodaService.get_by_id(db, doda_id, int(tenant_id), company_id)
if doda_record:
try:
DodaAltaLogService.create_from_alta_result(
db=db,
doda=doda_record,
company_id=company_id,
tenant_id=int(tenant_id),
variant=variant,
ext_result=ext_result,
action="consulta",
)
except Exception:
logger.exception("Error persistiendo DodaAltaLog(consulta) para doda_id=%s", doda_id)
return ext_result
@router.get(
"/consulta-status/{task_id}",
summary="Consultar estado de tarea de Consulta DODA",
tags=["doda-alta"],
)
async def get_doda_consulta_status(
task_id: str,
current_user: dict = Depends(get_current_user),
) -> Any:
try:
ext = DodaExternalService()
return ext.get_consulta_status(task_id)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=str(exc),
) from exc
except Exception as exc:
logger.exception("Error consultando consulta-status DODA task_id=%s", task_id)
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"Error al consultar el estado de la consulta DODA: {exc}",
) from exc
@router.post(
"/{doda_id}/consulta-apply/{task_id}",
summary="Aplicar resultado exitoso de consulta DODA al registro local",
tags=["doda-alta"],
)
async def post_doda_consulta_apply(
doda_id: int,
task_id: str,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
) -> Dict[str, Any]:
tenant_id = int(validate_access_to_resource(db, company_id, current_user))
doda_record = DodaService.get_by_id(db, doda_id, tenant_id, company_id)
if not doda_record:
raise HTTPException(status_code=404, detail="DODA no encontrado.")
try:
ext = DodaExternalService()
status_payload = ext.get_consulta_status(task_id)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=str(exc),
) from exc
except Exception as exc:
logger.exception("Error al consultar consulta-status para aplicar: task_id=%s", task_id)
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"Error al consultar el estado de la consulta DODA: {exc}",
) from exc
task_state = str(status_payload.get("state") or status_payload.get("status") or "").upper()
if task_state != "SUCCESS":
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="La tarea de consulta aún no está en estado SUCCESS.",
)
try:
_apply_consulta_success_to_doda(doda_record, status_payload)
if not (doda_record.status or "").strip():
doda_record.status = "VALIDADO"
db.add(doda_record)
db.commit()
db.refresh(doda_record)
except Exception as exc:
db.rollback()
logger.exception("Error aplicando consulta-status al DODA id=%s", doda_id)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"No se pudo aplicar el resultado de consulta al DODA: {exc}",
) from exc
return {
"message": "Resultado de consulta aplicado correctamente.",
"doda_id": doda_id,
"task_id": task_id,
"state": task_state,
}
@router.post(
"/{doda_id}/eliminar",
summary="Enviar Eliminación DODA al servicio externo (asíncrono)",
tags=["doda-alta"],
)
async def post_doda_eliminar(
doda_id: int,
company_id: int = Query(..., description="Company ID"),
variant: str = Query("doda", description="Tipo de alta: 'doda' o 'pita'"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
) -> Dict[str, Any]:
tenant_id = validate_access_to_resource(db, company_id, current_user)
user_email = (
current_user.get("email")
or current_user.get("preferred_username")
or ""
)
service = DodaAltaService(db)
try:
payload = service.build_eliminar_payload(
doda_id=doda_id,
tenant_id=int(tenant_id),
company_id=company_id,
variant=variant,
user_email=user_email,
)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=str(exc),
) from exc
ext_result: Dict[str, Any]
try:
ext = DodaExternalService()
ext_result = ext.post_eliminar(payload)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=str(exc),
) from exc
except Exception as exc:
logger.exception("Error al enviar eliminación DODA al servicio externo: doda_id=%s", doda_id)
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"Error al contactar el servicio DODA externo (eliminación): {exc}",
) from exc
doda_record = DodaService.get_by_id(db, doda_id, int(tenant_id), company_id)
if doda_record:
try:
DodaAltaLogService.create_from_alta_result(
db=db,
doda=doda_record,
company_id=company_id,
tenant_id=int(tenant_id),
variant=variant,
ext_result=ext_result,
action="eliminar",
)
except Exception:
logger.exception("Error persistiendo DodaAltaLog(eliminar) para doda_id=%s", doda_id)
try:
doda_record.integration_number = None
doda_record.transaction_number = None
doda_record.status = "PENDIENTE"
doda_record.sat_digital_seal = None
doda_record.sat_certificate = None
doda_record.serial_number = None
doda_record.electronic_signature = None
doda_record.original_chain = None
doda_record.sat_original_chain = None
doda_record.linq_sat_qr = None
doda_record.xml_doda_sent_path = None
doda_record.xml_doda_response_path = None
db.add(doda_record)
db.commit()
except Exception:
db.rollback()
logger.exception("Error desprocesando DODA local tras eliminación id=%s", doda_id)
return ext_result
@router.get(
"/eliminar-status/{task_id}",
summary="Consultar estado de tarea de Eliminación DODA",
tags=["doda-alta"],
)
async def get_doda_eliminar_status(
task_id: str,
current_user: dict = Depends(get_current_user),
) -> Any:
try:
ext = DodaExternalService()
return ext.get_eliminar_status(task_id)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=str(exc),
) from exc
except Exception as exc:
logger.exception("Error consultando eliminar-status DODA task_id=%s", task_id)
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"Error al consultar el estado de la eliminación DODA: {exc}",
) from exc
# ============ DODA ALTA LOG CRUD ============

30
backend/api/v1/modules/a76/general_catalogs/doda/service.py
View File

@@ -37,6 +37,19 @@ logger = logging.getLogger(__name__)
class DodaService:
"""Servicio para gestión de DODA"""
@staticmethod
def _ensure_editable_doda(doda: Optional[Doda]) -> None:
if not doda:
return
if (doda.integration_number or "").strip():
raise HTTPException(
status_code=422,
detail=(
"El DODA ya fue generado (tiene número de integración). "
"Elimínelo primero para poder editarlo."
),
)
@staticmethod
def _invalidate_report_after_mutation(
db: Session,
@@ -132,6 +145,7 @@ class DodaService:
db_doda = DodaService.get_by_id(db, id, tenant_id, company_id)
if not db_doda:
return None
DodaService._ensure_editable_doda(db_doda)
for key, value in doda_data.model_dump(exclude_unset=True).items():
setattr(db_doda, key, value)
@@ -195,6 +209,7 @@ class DodaService:
doda = db.query(Doda).filter(Doda.id == doda_id).first()
if not doda:
return None
DodaService._ensure_editable_doda(doda)
cv = (container_data.container_value or "").strip()
if not cv:
@@ -259,6 +274,8 @@ class DodaService:
)
if not db_container:
return None
doda = db.get(Doda, doda_id)
DodaService._ensure_editable_doda(doda)
for key, value in container_data.model_dump(exclude_unset=True).items():
setattr(db_container, key, value)
@@ -308,6 +325,8 @@ class DodaService:
)
if not db_container:
raise HTTPException(status_code=404, detail="Contenedor no encontrado.")
doda = db.get(Doda, doda_id)
DodaService._ensure_editable_doda(doda)
has_seals = bool(db_container.seals_detail)
if not has_seals and db_container.seals:
@@ -376,6 +395,7 @@ class DodaService:
doda = db.query(Doda).filter(Doda.id == doda_id).first()
if not doda:
raise HTTPException(status_code=404, detail="DODA no encontrado.")
DodaService._ensure_editable_doda(doda)
container = (
db.query(DodaContainer)
@@ -387,6 +407,8 @@ class DodaService:
)
if not container:
raise HTTPException(status_code=404, detail="Contenedor no encontrado.")
doda = db.get(Doda, doda_id)
DodaService._ensure_editable_doda(doda)
raw_value = (seal_data.seal_value or "").strip()
if not raw_value:
@@ -504,6 +526,7 @@ class DodaService:
doda = db.query(Doda).filter(Doda.id == doda_id).first()
if not doda:
return None
DodaService._ensure_editable_doda(doda)
tipo = (pedimento_data.american_pedimento_type or "").strip()
valor = (pedimento_data.american_pedimento_value or "").strip()
@@ -595,8 +618,9 @@ class DodaService:
raise HTTPException(
status_code=404, detail="Pedimento americano no encontrado."
)
doda = db.get(Doda, doda_id)
DodaService._ensure_editable_doda(doda)
try:
doda = db.get(Doda, doda_id)
db.delete(db_pedimento)
db.commit()
if doda:
@@ -625,6 +649,7 @@ class DodaService:
doda = db.query(Doda).filter(Doda.id == doda_id).first()
if not doda:
return None
DodaService._ensure_editable_doda(doda)
max_line = (
db.query(DodaPedimento)
@@ -681,8 +706,9 @@ class DodaService:
)
if not db_pedimento:
raise HTTPException(status_code=404, detail="Pedimento no encontrado.")
doda = db.get(Doda, doda_id)
DodaService._ensure_editable_doda(doda)
try:
doda = db.get(Doda, doda_id)
db.delete(db_pedimento)
db.commit()
if doda:

9
backend/api/v1/modules/a76/general_catalogs/electronic_notices/routes.py
View File

@@ -28,6 +28,11 @@ router = TenantCRUDRoutes(
resource_name="Electronic Notice",
enable_list=True,
enable_filters=True,
list_permissions=["cat_notices.view"],
get_permissions=["cat_notices.view"],
create_permissions=["cat_notices.create"],
update_permissions=["cat_notices.edit"],
delete_permissions=["cat_notices.delete"],
).router
@@ -43,7 +48,7 @@ async def get_notices_by_pedimento(
current_user: dict = Depends(get_current_user),
):
"""Get all electronic notices for a specific pedimento"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_notices.view"])
notices = ElectronicNoticeService.get_by_pedimento(
db, pedimento, tenant_id, company_id)
return [ElectronicNoticeResponseDTO.model_validate(notice) for notice in notices]
@@ -61,7 +66,7 @@ async def get_notices_by_status(
current_user: dict = Depends(get_current_user),
):
"""Get all electronic notices with a specific status"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_notices.view"])
notices = ElectronicNoticeService.get_by_status(
db, status, tenant_id, company_id)
return [ElectronicNoticeResponseDTO.model_validate(notice) for notice in notices]

101
backend/api/v1/modules/a76/general_catalogs/fractions/tariff_fractions/service.py
View File

@@ -9,6 +9,7 @@ from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
import zlib
import logging
import re
from .models import TariffFraction
from .dto import TariffFractionCreateDTO, TariffFractionUpdateDTO
@@ -22,7 +23,42 @@ logger = logging.getLogger(__name__)
class TariffFractionMapper:
"""Helper to map Sitar responses to Local domain objects"""
@staticmethod
def _digits_only(value: Optional[str]) -> str:
return re.sub(r"\D", "", (value or "").strip())
@staticmethod
def _format_mx_fraction(code: str) -> str:
if code.isdigit() and len(code) == 8:
return f"{code[:2]}.{code[2:4]}.{code[4:6]}.{code[6:]}"
if code.isdigit() and len(code) == 6:
return f"{code[:2]}.{code[2:4]}.{code[4:]}"
return code
@staticmethod
def _format_usa_fraction(code: str) -> str:
if code.isdigit() and len(code) == 10:
return f"{code[:4]}.{code[4:6]}.{code[6:8]}.{code[8:]}"
if code.isdigit() and len(code) == 8:
return f"{code[:4]}.{code[4:6]}.{code[6:]}"
return code
@staticmethod
def _normalized_pair(raw_code: Optional[str], raw_fraction: Optional[str], formatter) -> Tuple[str, str]:
"""Return (code_without_separators, formatted_fraction)."""
code = TariffFractionMapper._digits_only(raw_code)
fraction = (raw_fraction or "").strip()
if not code:
code = TariffFractionMapper._digits_only(fraction)
if not fraction:
fraction = formatter(code)
elif "." not in fraction and "-" not in fraction:
fraction = formatter(TariffFractionMapper._digits_only(fraction))
if not fraction:
fraction = formatter(code)
return code, fraction
@staticmethod
def to_domain(fraccion: FraccionesResponse) -> TariffFraction:
# Generate ID: Use SYSID if available, else composite hash of code + nico
@@ -33,22 +69,9 @@ class TariffFractionMapper:
unique_str = f"{fraccion.FRACCION}-{fraccion.NICO}"
fake_id = zlib.crc32(unique_str.encode('utf-8'))
# UX Enhauncement: Sitar API returns empty strings for some fields.
# We fill them with fallbacks so the frontend table isn't 90% empty.
code_val = fraccion.FRACCION
# Formatting Logic: if FRACCIONPUNTO is empty, try to format code_val
formatted_fraction = code_val
if fraccion.FRACCIONPUNTO:
formatted_fraction = fraccion.FRACCIONPUNTO
elif code_val and code_val.isdigit() and len(code_val) == 8:
# Standard 8 digit format: XX.XX.XX.XX
formatted_fraction = f"{code_val[:2]}.{code_val[2:4]}.{code_val[4:6]}.{code_val[6:]}"
elif code_val and code_val.isdigit() and len(code_val) == 6:
# 6 digit (subheading): XX.XX.XX
formatted_fraction = f"{code_val[:2]}.{code_val[2:4]}.{code_val[4:]}"
fraction_val = formatted_fraction
code_val, fraction_val = TariffFractionMapper._normalized_pair(
fraccion.FRACCION, fraccion.FRACCIONPUNTO, TariffFractionMapper._format_mx_fraction
)
description_val = fraccion.DESCRIPCION if fraccion.DESCRIPCION else "(Sin descripción)"
tf = TariffFraction(
@@ -73,10 +96,15 @@ class TariffFractionMapper:
@staticmethod
def to_domain_usa(item: FraccionesUSAResponse) -> TariffFraction:
"""Map US Fraction to Domain"""
code_val, fraction_val = TariffFractionMapper._normalized_pair(
item.FRACCION_SIN_PUNTO,
item.FRACCION_CON_PUNTO or item.FRACCION_MOSTRAR,
TariffFractionMapper._format_usa_fraction,
)
return TariffFraction(
id=item.CONSECUTIVO,
code=item.FRACCION_SIN_PUNTO or "",
fraction=item.FRACCION_CON_PUNTO or "",
code=code_val,
fraction=fraction_val,
description=item.DESCRIPCION or "(Sin descripción)",
nico=None, # Not applicable
umt=item.UNIDADCANTIDAD,
@@ -183,28 +211,27 @@ class TariffFractionService:
# Map filters
sitar_fraccion = None
sitar_nico = None
sitar_description = None
# Default level logic
level_filter = 5 # Default legacy
# Legacy parity: base query is always Nivel = 5 unless caller explicitly requests another level.
level_filter = 5
if filters and filters.get("level") is not None:
level_filter = filters["level"]
# Allow disabling level filter explicitly
# UI compatibility: level -1 means "sin filtro de nivel".
if level_filter == -1:
level_filter = None
if filters:
if filters.get("search"):
term = filters["search"]
# Heuristic: if search starts with digit (after removing dots), treat as code/fraccion/nico
# This covers "0101", "01.01", "020691A"
term = str(filters["search"]).strip()
# Legacy-like behavior:
# - Numeric search targets fracción first.
# - Text search targets descripción.
clean_term = term.replace(".", "")
if clean_term and clean_term[0].isdigit():
if clean_term.isdigit():
sitar_fraccion = clean_term
else:
# Attempt description search via API first
logger.info(f"Search term '{term}' identified as text. Attempting API description search.")
pass
sitar_description = term
if filters.get("code"):
sitar_fraccion = filters["code"]
@@ -212,14 +239,8 @@ class TariffFractionService:
sitar_fraccion = filters["fraction"]
if filters.get("nico"):
sitar_nico = filters["nico"]
# Determine description filter
sitar_description = None
# Only use description if we didn't use it as code above
if filters and filters.get("search"):
clean_term = filters["search"].replace(".", "")
if not (clean_term and clean_term[0].isdigit()):
sitar_description = filters["search"]
if sitar_fraccion is not None:
sitar_fraccion = str(sitar_fraccion).replace(".", "").strip()
# Note: Sitar search might not return total count.
# We fetch page items. Pagination might be tricky if Sitar doesn't return total.
@@ -239,6 +260,8 @@ class TariffFractionService:
# Map items
items = [TariffFractionMapper.to_domain(item) for item in sitar_items]
# Legacy browse behavior: keep table in ascending fracción order.
items = sorted(items, key=lambda row: ((row.code or ""), (row.nico or "")))
# Estimate total (Sitar service doesn't return total currently)
# If we got full limit, assume there are more.
@@ -289,8 +312,6 @@ class TariffFractionService:
query = query.filter(TariffFraction.umt.ilike(f"%{filters['umt']}%"))
total = query.count()
# Add deterministic sort order
query = query.order_by(TariffFraction.fraction)
items = query.offset(skip).limit(limit).all()
return items, total

19
backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/dto.py
View File

@@ -6,6 +6,7 @@ from datetime import datetime
from typing import Optional, Any
from pydantic import BaseModel, Field, ConfigDict, model_validator
import re
class USTariffFractionCreateDTO(BaseModel):
@@ -62,10 +63,20 @@ class USTariffFractionResponseDTO(BaseModel):
if raw_code:
code_str = str(raw_code)
# fraction keeps the original formatted string
fraction = code_str
# code strips dots and hyphens
code = code_str.replace(".", "").replace("-", "")
fraction_raw = ""
if isinstance(data, dict):
fraction_raw = str(data.get("fraction") or "")
else:
fraction_raw = str(getattr(data, "fraction", "") or "")
code = re.sub(r"[.\s-]", "", code_str)
fraction = fraction_raw.strip() or code_str
if "." not in fraction and "-" not in fraction:
only_digits = re.sub(r"[.\s-]", "", fraction)
if len(only_digits) == 10:
fraction = f"{only_digits[:4]}.{only_digits[4:6]}.{only_digits[6:8]}.{only_digits[8:]}"
elif len(only_digits) == 8:
fraction = f"{only_digits[:4]}.{only_digits[4:6]}.{only_digits[6:]}"
if isinstance(data, dict):
data["code"] = code

1
backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/routes.py
View File

@@ -28,6 +28,7 @@ def _sitar_row_to_us_response_payload(item: FraccionesUSAResponse) -> dict:
return {
"id": item.CONSECUTIVO,
"code": canon,
"fraction": item.FRACCION_CON_PUNTO or item.FRACCION_MOSTRAR or canon,
"prefix": item.FRACCION_SIN_PUNTO,
"type_code": str(item.NIVEL) if item.NIVEL is not None else None,
"ad_valorem": american_fraction_ad_valorem_from_row(item),

2
backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/service.py
View File

@@ -128,7 +128,7 @@ class USTariffFractionService:
)
total = query.count()
items = query.order_by(USTariffFraction.code).offset(skip).limit(limit).all()
items = query.offset(skip).limit(limit).all()
return items, total

1
backend/api/v1/modules/a76/general_catalogs/identifiers/routes.py
View File

@@ -19,6 +19,7 @@ identifier_crud = TenantCRUDRoutes(
tags=["Identifiers"],
resource_name="Identifier",
enable_list=True,
enable_filters=True,
list_permissions=["cat_identifiers.view"],
get_permissions=["cat_identifiers.view"],
create_permissions=["cat_identifiers.create"],

4
backend/api/v1/modules/a76/invoices/services.py
View File

@@ -1,7 +1,7 @@
import traceback
from typing import Optional, List, Tuple
from sqlalchemy.orm import Session
from sqlalchemy import func
from sqlalchemy import and_, func, or_
from core.exceptions import ErrorCollector, DuplicateResourceException
from core.context import get_user_context
from .common.mappers import clean_dict
@@ -328,7 +328,6 @@ class InvoiceService:
# Filtro por permisos granulares (allowed_types)
if "allowed_types" in filters:
from sqlalchemy import or_, and_
allowed = filters["allowed_types"]
if allowed is None:
# Acceso global (admin o view_all) - no filtramos por tipos
@@ -337,7 +336,6 @@ class InvoiceService:
# Seguridad: Si el usuario NO tiene permisos para ningún tipo específico
query = query.filter(models.InvoiceHeader.id == -1)
else:
from sqlalchemy import func
conditions = []
for op, inv in allowed:
# Aseguramos comparación insensible a mayúsculas para mayor robustez con la DB

1
backend/api/v1/modules/a76/layouts_csv/parts/tasks.py
View File

@@ -325,7 +325,6 @@ def insert_valid_rows(self, job_id: str):
new_part = Part(
tenant_id=tenant_id,
company_id=company_id,
client_id=company_id,
**data,
)
session.add(new_part)

2
backend/api/v1/modules/a76/parts/dto.py
View File

@@ -145,7 +145,6 @@ class InvDataDTO(BaseModel):
class PartBase(BaseModel):
client_id: Optional[int] = None
part_number: str = Field(..., max_length=70)
commercial_part_number: Optional[str] = None
@@ -186,7 +185,6 @@ class PartCreateDTO(PartBase):
# --- ACTUALIZACIÓN ---
class PartUpdateDTO(PartBase):
client_id: Optional[int] = None
part_number: Optional[str] = None
# Todo opcional para PATCH
pass

1
backend/api/v1/modules/a76/parts/models.py
View File

@@ -68,7 +68,6 @@ class Part(Base, TenantScopedMixin, TimestampMixin):
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
client_id: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
part_number: Mapped[str] = mapped_column(String(70))
commercial_part_number: Mapped[Optional[str]] = mapped_column(String(70))

3
backend/api/v1/modules/public/reference_data/code_pedimento_regimens/routes.py
View File

@@ -16,7 +16,6 @@ router = APIRouter(prefix="/code-pedimento-regimens")
def list_code_pedimento_regimens(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=1000, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
code: str = Query(None, description="Filter by code"),
regime: str = Query(None, description="Filter by regime"),
type: str = Query(None, description="Filter by type"),
@@ -24,8 +23,6 @@ def list_code_pedimento_regimens(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_pedimento_regimens.view"])
skip = (page - 1) * page_size
query = db.query(CodePedimentoRegimen)

30
backend/api/v1/modules/public/reference_data/containers/routes.py
View File

@@ -1,7 +1,7 @@
from typing import Any, Dict
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from core.security import get_current_user, has_role
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy import or_
@@ -16,14 +16,10 @@ router = APIRouter(prefix="/containers")
async def list_containers(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
# 🛡️ Permiso de Lectura (Listado)
validate_access_to_resource(db, company_id, current_user, required_permissions=["ref_containers.view"])
skip = (page - 1) * page_size
query = db.query(Container)
@@ -48,13 +44,9 @@ async def list_containers(
@router.get("/{key}", response_model=ContainerDTO)
async def get_container(
key: str,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
# 🛡️ Permiso de Lectura (Individual)
validate_access_to_resource(db, company_id, current_user, required_permissions=["ref_containers.view"])
obj = db.query(Container).filter(Container.key == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -64,13 +56,9 @@ async def get_container(
@router.post("/", response_model=ContainerDTO, status_code=201)
async def create_container(
data: ContainerDTO,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
# 🛡️ Permiso de Creación
validate_access_to_resource(db, company_id, current_user, required_permissions=["ref_containers.create"])
obj = Container(**data.dict())
db.add(obj)
db.commit()
@@ -82,13 +70,9 @@ async def create_container(
async def update_container(
key: str,
data: ContainerDTO,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
# 🛡️ Permiso de Edición
validate_access_to_resource(db, company_id, current_user, required_permissions=["ref_containers.edit"])
obj = db.query(Container).filter(Container.key == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -102,16 +86,12 @@ async def update_container(
@router.delete("/{key}", status_code=204)
async def delete_container(
key: str,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
# 🛡️ Permiso de Borrado
validate_access_to_resource(db, company_id, current_user, required_permissions=["ref_containers.delete"])
obj = db.query(Container).filter(Container.key == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
db.delete(obj)
db.commit()
return None
return None

22
backend/api/v1/modules/public/reference_data/countries/routes.py
View File

@@ -1,7 +1,7 @@
from typing import Any, Dict
from core.database import get_core_db
from core.security import get_current_user
from core.security import get_current_user, has_role
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
@@ -15,14 +15,10 @@ router = APIRouter(prefix="/countries")
async def list_countries(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_countries.view", "cat_countries.view"], require_all=False)
"""Endpoint público para obtener lista de países - no requiere autenticación"""
skip = (page - 1) * page_size
query = db.query(Country)
@@ -53,8 +49,6 @@ async def get_country(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_countries", "view")
obj = db.query(Country).filter(Country.m3_key == m3_key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -65,10 +59,8 @@ async def get_country(
async def create_country(
data: CountryDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_countries", "create")
obj = Country(**data.dict())
db.add(obj)
db.commit()
@@ -81,10 +73,8 @@ async def update_country(
m3_key: str,
data: CountryDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_countries", "edit")
obj = db.query(Country).filter(Country.m3_key == m3_key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -99,10 +89,8 @@ async def update_country(
async def delete_country(
m3_key: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_countries", "delete")
obj = db.query(Country).filter(Country.m3_key == m3_key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")

19
backend/api/v1/modules/public/reference_data/currency_types/routes.py
View File

@@ -1,7 +1,7 @@
from typing import Any, Dict
from core.database import get_core_db
from core.security import get_current_user
from core.security import get_current_user, has_role
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy import or_
@@ -16,13 +16,10 @@ router = APIRouter(prefix="/currency-types")
async def list_currency_types(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_currency_types.view", "cat_currency.view"], require_all=False)
skip = (page - 1) * page_size
query = db.query(CurrencyType)
@@ -51,8 +48,6 @@ async def get_currency_type(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_currency_types", "view")
obj = db.query(CurrencyType).filter(CurrencyType.code == code).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -63,10 +58,8 @@ async def get_currency_type(
async def create_currency_type(
data: CurrencyTypeDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_currency_types", "create")
obj = CurrencyType(**data.dict())
db.add(obj)
db.commit()
@@ -79,10 +72,8 @@ async def update_currency_type(
code: str,
data: CurrencyTypeDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_currency_types", "edit")
obj = db.query(CurrencyType).filter(CurrencyType.code == code).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -97,10 +88,8 @@ async def update_currency_type(
async def delete_currency_type(
code: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_currency_types", "delete")
obj = db.query(CurrencyType).filter(CurrencyType.code == code).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")

21
backend/api/v1/modules/public/reference_data/customs_sections/routes.py
View File

@@ -16,13 +16,10 @@ router = APIRouter(prefix="/customs-sections")
def list_customs_sections(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_customs_sections.view"])
skip = (page - 1) * page_size
query = db.query(CustomsSection)
@@ -47,12 +44,9 @@ def list_customs_sections(
@router.get("/{customs_code}", response_model=CustomsSectionDTO)
def get_customs_section(
customs_code: str,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_customs_sections.view"])
obj = (
db.query(CustomsSection)
.filter(CustomsSection.customs_code == customs_code)
@@ -66,12 +60,9 @@ def get_customs_section(
@router.post("/", response_model=CustomsSectionDTO, status_code=201)
def create_customs_section(
data: CustomsSectionDTO,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_customs_sections.edit"])
obj = CustomsSection(**data.dict())
db.add(obj)
db.commit()
@@ -83,12 +74,9 @@ def create_customs_section(
def update_customs_section(
customs_code: str,
data: CustomsSectionDTO,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_customs_sections.edit"])
obj = (
db.query(CustomsSection)
.filter(CustomsSection.customs_code == customs_code)
@@ -106,12 +94,9 @@ def update_customs_section(
@router.delete("/{customs_code}", status_code=204)
def delete_customs_section(
customs_code: str,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_customs_sections.edit"])
obj = (
db.query(CustomsSection)
.filter(CustomsSection.customs_code == customs_code)

3
backend/api/v1/modules/public/reference_data/customs_warehouses/routes.py
View File

@@ -16,13 +16,10 @@ router = APIRouter(prefix="/customs-warehouses")
def list_customs_warehouses(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_customs_warehouses.view", "cat_warehouses.view"], require_all=False)
skip = (page - 1) * page_size
query = db.query(CustomsWarehouse)

18
backend/api/v1/modules/public/reference_data/identifiers/routes.py
View File

@@ -1,7 +1,7 @@
from typing import Any, Dict
from core.database import get_core_db
from core.security import get_current_user
from core.security import get_current_user, has_role
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import or_
from sqlalchemy.orm import Session
@@ -22,8 +22,6 @@ async def list_identifiers(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_identifiers", "view")
skip = (page - 1) * page_size
query = db.query(IdentifierCatalog)
@@ -67,8 +65,6 @@ async def get_identifier(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_identifiers", "view")
obj = db.query(IdentifierCatalog).filter(IdentifierCatalog.key == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -79,10 +75,8 @@ async def get_identifier(
async def create_identifier(
data: IdentifierDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_identifiers", "create")
# Check if already exists
existing = db.query(IdentifierCatalog).filter(IdentifierCatalog.key == data.key).first()
if existing:
@@ -100,10 +94,8 @@ async def update_identifier(
key: str,
data: IdentifierDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_identifiers", "edit")
obj = db.query(IdentifierCatalog).filter(IdentifierCatalog.key == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -118,10 +110,8 @@ async def update_identifier(
async def delete_identifier(
key: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource as validate_perm
validate_perm(current_user, "cat_identifiers", "delete")
obj = db.query(IdentifierCatalog).filter(IdentifierCatalog.key == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")

3
backend/api/v1/modules/public/reference_data/incoterms/routes.py
View File

@@ -16,15 +16,12 @@ router = APIRouter(prefix="/incoterms")
async def list_incoterms(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
code: str = Query(None, description="Filtrar por clave"),
description: str = Query(None, description="Filtrar por descripción"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_incoterms.view", "cat_incoterms.view"], require_all=False)
skip = (page - 1) * page_size
query = db.query(Incoterm)

4
backend/api/v1/modules/public/reference_data/invoice_types/routes.py
View File

@@ -16,15 +16,11 @@ router = APIRouter(prefix="/invoice-types")
def list_invoice_types(
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=100),
company_id: int = Query(..., description="ID de la empresa"),
type: Optional[str] = Query(None, description="Filter by type"),
operation: Optional[str] = Query(None, description="Filter by operation type (imp, exp, both)"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_invoice_types.view", "cat_inv_types.view"], require_all=False)
query = db.query(InvoiceType)
if search:

3
backend/api/v1/modules/public/reference_data/material_types/routes.py
View File

@@ -16,14 +16,11 @@ router = APIRouter(prefix="/material-types")
async def list_material_types(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=1000, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
type: str = Query(None, description="Filtrar por tipo (ACTIVO FIJO, MATERIALES, PRODUCTOS)"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_material_types.view", "cat_material_types.view"], require_all=False)
skip = (page - 1) * page_size
query = db.query(MaterialType)

6
backend/api/v1/modules/public/reference_data/payment_methods/routes.py
View File

@@ -16,13 +16,10 @@ router = APIRouter(prefix="/payment-methods")
def list_payment_methods(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["pedimentos_payments.view"])
skip = (page - 1) * page_size
query = db.query(PaymentMethod)
@@ -47,12 +44,9 @@ def list_payment_methods(
@router.get("/{key}", response_model=PaymentMethodDTO)
def get_payment_method(
key: str,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["pedimentos_payments.view"])
obj = db.query(PaymentMethod).filter(PaymentMethod.key == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")

6
backend/api/v1/modules/public/reference_data/pedimento_codes/routes.py
View File

@@ -16,13 +16,10 @@ router = APIRouter(prefix="/pedimento-codes")
def list_pedimento_codes(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=1000, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_pedimento_codes.view"])
skip = (page - 1) * page_size
query = db.query(PedimentoCode)
@@ -47,12 +44,9 @@ def list_pedimento_codes(
@router.get("/{code}", response_model=PedimentoCodeDTO)
def get_pedimento_code(
code: str,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_pedimento_codes .view"])
obj = db.query(PedimentoCode).filter(PedimentoCode.code == code).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")

24
backend/api/v1/modules/public/reference_data/pedimento_regimens/routes.py
View File

@@ -1,7 +1,7 @@
from typing import Any, Dict
from core.database import get_core_db
from core.security import get_current_user
from core.security import get_current_user, has_role
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy import or_
@@ -16,13 +16,10 @@ router = APIRouter(prefix="/pedimento-regimens")
def list_pedimento_regimens(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_pedimento_regimens.view"])
skip = (page - 1) * page_size
query = db.query(RegimenPedimento)
@@ -47,12 +44,9 @@ def list_pedimento_regimens(
@router.get("/{key}", response_model=RegimenPedimentoDTO)
def get_pedimento_regimen(
key: str,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_pedimento_regimens.view"])
obj = db.query(RegimenPedimento).filter(RegimenPedimento.code == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -62,13 +56,9 @@ def get_pedimento_regimen(
@router.post("/", response_model=RegimenPedimentoDTO, status_code=201)
def create_pedimento_regimen(
data: RegimenPedimentoDTO,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource
# Mutation for reference data usually restricted to admin role or specific perm
validate_access_to_resource(db, company_id, current_user, ["ref_pedimento_regimens.edit"])
obj = RegimenPedimento(**data.model_dump())
db.add(obj)
db.commit()
@@ -80,12 +70,9 @@ def create_pedimento_regimen(
def update_pedimento_regimen(
key: str,
data: RegimenPedimentoDTO,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_pedimento_regimens.edit"])
obj = db.query(RegimenPedimento).filter(RegimenPedimento.code == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -99,12 +86,9 @@ def update_pedimento_regimen(
@router.delete("/{key}", status_code=204)
def delete_pedimento_regimen(
key: str,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
current_user: dict = Depends(has_role("admin")),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_pedimento_regimens.edit"])
obj = db.query(RegimenPedimento).filter(RegimenPedimento.code == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")

28
backend/api/v1/modules/public/reference_data/pedimento_transport_catalog/routes.py
View File

@@ -14,15 +14,11 @@ router = APIRouter(prefix="/pedimento-transport-catalog")
@router.get("/", response_model=Dict[str, Any])
async def list_pedimento_transport_catalog(
company_id: int = Query(..., description="ID de la empresa"),
page: int = Query(1, ge=1, description="Numero de pagina"),
page_size: int = Query(100, ge=1, le=200, description="Tamano de pagina"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["pedimentos_anexo22.view"])
skip = (page - 1) * page_size
query = db.query(PedimentoTransportCatalog)
@@ -47,14 +43,7 @@ async def list_pedimento_transport_catalog(
@router.get("/{code}", response_model=PedimentoTransportCatalogDTO)
async def get_pedimento_transport_catalog(
code: str,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["pedimentos_anexo22.view"])
async def get_pedimento_transport_catalog(code: str, db: Session = Depends(get_core_db)):
obj = (
db.query(PedimentoTransportCatalog)
.filter(PedimentoTransportCatalog.code == code)
@@ -68,12 +57,9 @@ async def get_pedimento_transport_catalog(
@router.post("/", response_model=PedimentoTransportCatalogDTO, status_code=201)
async def create_pedimento_transport_catalog(
data: PedimentoTransportCatalogDTO,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
user=Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["pedimentos_anexo22.view"])
obj = PedimentoTransportCatalog(**data.model_dump())
db.add(obj)
db.commit()
@@ -85,12 +71,9 @@ async def create_pedimento_transport_catalog(
async def update_pedimento_transport_catalog(
code: str,
data: PedimentoTransportCatalogDTO,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
user=Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["pedimentos_anexo22.view"])
obj = (
db.query(PedimentoTransportCatalog)
.filter(PedimentoTransportCatalog.code == code)
@@ -108,12 +91,9 @@ async def update_pedimento_transport_catalog(
@router.delete("/{code}", status_code=204)
async def delete_pedimento_transport_catalog(
code: str,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
user=Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["pedimentos_anexo22.view"])
obj = (
db.query(PedimentoTransportCatalog)
.filter(PedimentoTransportCatalog.code == code)

3
backend/api/v1/modules/public/reference_data/states/routes.py
View File

@@ -16,13 +16,10 @@ router = APIRouter(prefix="/states")
async def list_states(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_states.view"])
skip = (page - 1) * page_size
query = db.query(State)

39
backend/api/v1/modules/public/reference_data/transport_modes/routes.py
View File

@@ -1,7 +1,7 @@
from typing import Any, Dict
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from core.security import get_current_user
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy import or_
@@ -16,14 +16,9 @@ router = APIRouter(prefix="/transport-modes")
async def list_transport_modes(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
# 🛡️ Permiso de Lectura (Listado)
validate_access_to_resource(db, company_id, current_user, required_permissions=["ref_transport_modes.view"])
skip = (page - 1) * page_size
query = db.query(TransportMode)
@@ -46,15 +41,7 @@ async def list_transport_modes(
@router.get("/{key}", response_model=TransportModeDTO)
async def get_transport_mode(
key: str,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
# 🛡️ Permiso de Lectura (Individual)
validate_access_to_resource(db, company_id, current_user, required_permissions=["ref_transport_modes.view"])
async def get_transport_mode(key: str, db: Session = Depends(get_core_db)):
obj = db.query(TransportMode).filter(TransportMode.key == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -64,13 +51,9 @@ async def get_transport_mode(
@router.post("/", response_model=TransportModeDTO, status_code=201)
async def create_transport_mode(
data: TransportModeDTO,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
user=Depends(get_current_user),
):
# 🛡️ Permiso de Creación
validate_access_to_resource(db, company_id, current_user, required_permissions=["ref_transport_modes.create"])
obj = TransportMode(**data.dict())
db.add(obj)
db.commit()
@@ -82,13 +65,9 @@ async def create_transport_mode(
async def update_transport_mode(
key: str,
data: TransportModeDTO,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
user=Depends(get_current_user),
):
# 🛡️ Permiso de Edición
validate_access_to_resource(db, company_id, current_user, required_permissions=["ref_transport_modes.edit"])
obj = db.query(TransportMode).filter(TransportMode.key == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -101,17 +80,11 @@ async def update_transport_mode(
@router.delete("/{key}", status_code=204)
async def delete_transport_mode(
key: str,
company_id: int = Query(..., description="ID de la empresa"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
key: str, db: Session = Depends(get_core_db), user=Depends(get_current_user)
):
# 🛡️ Permiso de Borrado
validate_access_to_resource(db, company_id, current_user, required_permissions=["ref_transport_modes.delete"])
obj = db.query(TransportMode).filter(TransportMode.key == key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
db.delete(obj)
db.commit()
return None
return None

4
backend/api/v1/modules/public/reference_data/transport_types/routes.py
View File

@@ -16,13 +16,9 @@ router = APIRouter(prefix="/transport-types")
def list_transport_types(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_transport_types.view"])
skip = (page - 1) * page_size
query = db.query(TransportType)

3
backend/api/v1/modules/public/reference_data/valuation_methods/routes.py
View File

@@ -16,13 +16,10 @@ router = APIRouter(prefix="/valuation-methods")
async def list_valuation_methods(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
company_id: int = Query(..., description="ID de la empresa"),
search: str = Query(None, description="Término de búsqueda"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
from core.security import validate_access_to_resource
validate_access_to_resource(db, company_id, current_user, ["ref_valuation_methods.view", "cat_valuation.view"], require_all=False)
skip = (page - 1) * page_size
query = db.query(ValuationMethod)

69
backend/tests/unit/general_catalogs/doda/test_doda_alta_payloads.py Normal file
View File

@@ -0,0 +1,69 @@
from __future__ import annotations
import json
from types import SimpleNamespace
import pytest
from api.v1.modules.a76.general_catalogs.doda.alta_service import DodaAltaService
def test_build_consulta_payload_appends_numero_transaccion(monkeypatch):
service = DodaAltaService(db=None)
monkeypatch.setattr(
service,
"build_alta_payload",
lambda **kwargs: {"base": "payload"},
)
monkeypatch.setattr(
service,
"_latest_alta_log",
lambda *args, **kwargs: SimpleNamespace(
id=9,
result_json=json.dumps({"numero_transaccion": "TX-001"}),
integration_number="INT-001",
),
)
payload = service.build_consulta_payload(
doda_id=1, tenant_id=1, company_id=1, variant="doda", user_email="u@test.com"
)
assert payload["base"] == "payload"
assert payload["numero_transaccion"] == "TX-001"
def test_build_consulta_payload_fails_when_numero_transaccion_missing(monkeypatch):
service = DodaAltaService(db=None)
monkeypatch.setattr(service, "build_alta_payload", lambda **kwargs: {})
monkeypatch.setattr(
service,
"_latest_alta_log",
lambda *args, **kwargs: SimpleNamespace(
id=10, result_json=json.dumps({"other": "value"}), integration_number="INT-001"
),
)
with pytest.raises(ValueError, match="numero_transaccion"):
service.build_consulta_payload(
doda_id=1, tenant_id=1, company_id=1, variant="doda", user_email=""
)
def test_build_eliminar_payload_appends_numero_integracion(monkeypatch):
service = DodaAltaService(db=None)
monkeypatch.setattr(service, "build_alta_payload", lambda **kwargs: {"base": "payload"})
monkeypatch.setattr(
service,
"_latest_alta_log",
lambda *args, **kwargs: SimpleNamespace(
id=11,
result_json="{}",
integration_number="INT-900",
),
)
payload = service.build_eliminar_payload(
doda_id=2, tenant_id=1, company_id=1, variant="doda", user_email="user@test.com"
)
assert payload["base"] == "payload"
assert payload["numero_integracion"] == "INT-900"

65
backend/tests/unit/general_catalogs/doda/test_doda_external_service.py Normal file
View File

@@ -0,0 +1,65 @@
from __future__ import annotations
from typing import Any, Dict
from api.v1.modules.a76.general_catalogs.doda.external_service import DodaExternalService
class _FakeResponse:
def __init__(self, payload: Dict[str, Any]):
self._payload = payload
def raise_for_status(self) -> None:
return None
def json(self) -> Dict[str, Any]:
return self._payload
class _FakeClient:
calls = []
def __init__(self, *args, **kwargs):
self.kwargs = kwargs
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def post(self, url: str, json: Dict[str, Any]):
_FakeClient.calls.append(("POST", url, json))
return _FakeResponse({"task_id": "t-1", "status": "queued", "message": "ok"})
def get(self, url: str):
_FakeClient.calls.append(("GET", url, None))
return _FakeResponse({"task_id": "t-1", "status": "done", "message": "ok"})
def test_external_service_supports_consulta_and_eliminar(monkeypatch):
from api.v1.modules.a76.general_catalogs.doda import external_service as module
_FakeClient.calls = []
monkeypatch.setattr(module, "httpx", module.httpx)
monkeypatch.setattr(module.httpx, "Client", _FakeClient)
service = DodaExternalService()
service.base_url = "http://example.test"
payload = {"foo": "bar"}
consulta = service.post_consulta(payload)
consulta_status = service.get_consulta_status("abc123")
eliminar = service.post_eliminar(payload)
eliminar_status = service.get_eliminar_status("abc123")
assert consulta["task_id"] == "t-1"
assert consulta_status["status"] == "done"
assert eliminar["status"] == "queued"
assert eliminar_status["task_id"] == "t-1"
assert _FakeClient.calls == [
("POST", "http://example.test/api/v1/doda/consulta", payload),
("GET", "http://example.test/api/v1/doda/consulta-status/abc123", None),
("POST", "http://example.test/api/v1/doda/eliminar", payload),
("GET", "http://example.test/api/v1/doda/eliminar-status/abc123", None),
]

57
backend/tests/unit/general_catalogs/fractions/test_tariff_fraction_mapper.py Normal file
View File

@@ -0,0 +1,57 @@
from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.service import (
TariffFractionMapper,
)
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.dto import (
USTariffFractionResponseDTO,
)
from api.v1.modules.sitar.fracciones_usa.schemas import FraccionesUSAResponse
def test_to_domain_usa_keeps_separate_code_and_fraction():
row = FraccionesUSAResponse(
CONSECUTIVO=10,
FRACCION_SIN_PUNTO="1234567890",
FRACCION_CON_PUNTO="1234.56.78.90",
DESCRIPCION="Test",
UNIDADCANTIDAD="KG",
TARIFA1="5%",
TARIFA2="0%",
)
mapped = TariffFractionMapper.to_domain_usa(row)
assert mapped.code == "1234567890"
assert mapped.fraction == "1234.56.78.90"
def test_to_domain_usa_formats_fraction_when_only_code_available():
row = FraccionesUSAResponse(
CONSECUTIVO=11,
FRACCION_SIN_PUNTO="9876543210",
FRACCION_CON_PUNTO=None,
FRACCION_MOSTRAR=None,
DESCRIPCION="Fallback",
UNIDADCANTIDAD="PZA",
TARIFA1="7.5%",
TARIFA2="0%",
)
mapped = TariffFractionMapper.to_domain_usa(row)
assert mapped.code == "9876543210"
assert mapped.fraction == "9876.54.32.10"
def test_us_response_dto_preserves_fraction_when_provided():
dto = USTariffFractionResponseDTO.model_validate(
{
"id": 1,
"code": "1111.22.33.44",
"fraction": "1111.22.33.44",
"description": "DTO test",
"created_at": "2026-01-01T00:00:00Z",
"updated_at": "2026-01-01T00:00:00Z",
}
)
assert dto.code == "1111223344"
assert dto.fraction == "1111.22.33.44"