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

feature/catalogo-equivalencias

Browse Source
This commit is contained in:
hreyes
2026-03-18 09:40:52 -06:00
parent d37d6ef699
commit 8b9537b5a9
10 changed files with 379 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
backend/api/v1/modules/a76/general_catalogs/equivalencies/dto.py
View File

@@ -1,39 +1,15 @@
from typing import Optional, List
from typing import Optional
from decimal import Decimal
from pydantic import BaseModel, Field, ConfigDict
# Equivalency Item DTOs
class EquivalencyItemBase(BaseModel):
original_field: str = Field(..., max_length=100,
description="Original Field (Unit of Measure)")
external_field: str = Field(..., max_length=100,
description="External Field")
class EquivalencyItemCreate(EquivalencyItemBase):
pass
class EquivalencyItemUpdate(BaseModel):
original_field: Optional[str] = Field(None, max_length=100)
external_field: Optional[str] = Field(None, max_length=100)
class EquivalencyItemResponse(EquivalencyItemBase):
id: int
equivalency_id: int
model_config = ConfigDict(from_attributes=True)
# Equivalency DTOs
# Equivalency (parent) DTOs
class EquivalencyBase(BaseModel):
fraccion_mex: str = Field(..., max_length=10, description="Fraccion MX (Identifier)")
fraccion_us: str = Field(..., max_length=100, description="Fraccion US (External Field)")
description: Optional[str] = Field(
None, max_length=200, description="Description")
# Agrupa múltiples destinos para un mismo origen.
identifier: str = Field(..., max_length=10, description="From unit code")
description: Optional[str] = Field(None, max_length=200, description="Description (optional)")
class EquivalencyCreate(EquivalencyBase):
@@ -41,8 +17,7 @@ class EquivalencyCreate(EquivalencyBase):
class EquivalencyUpdate(BaseModel):
fraccion_mex: Optional[str] = Field(None, max_length=10)
fraccion_us: Optional[str] = Field(None, max_length=100)
identifier: Optional[str] = Field(None, max_length=10)
description: Optional[str] = Field(None, max_length=200)
@@ -53,4 +28,30 @@ class EquivalencyResponse(EquivalencyBase):
model_config = ConfigDict(from_attributes=True)
# Equivalency Item DTOs (catálogo por fila)
class EquivalencyItemBase(BaseModel):
original_field: str = Field(..., max_length=100, description="Original unit code (from)")
external_field: str = Field(..., max_length=100, description="External unit code (to)")
conversion_factor: Optional[Decimal] = Field(None, description="Conversion factor")
class EquivalencyItemCreate(EquivalencyItemBase):
conversion_factor: Decimal = Field(..., description="Conversion factor")
class EquivalencyItemUpdate(BaseModel):
original_field: Optional[str] = Field(None, max_length=100)
external_field: Optional[str] = Field(None, max_length=100)
conversion_factor: Optional[Decimal] = None
class EquivalencyItemResponse(EquivalencyItemBase):
id: int
equivalency_id: int
tenant_id: int
company_id: int
model_config = ConfigDict(from_attributes=True)

7
backend/api/v1/modules/a76/general_catalogs/equivalencies/models.py
View File

@@ -1,5 +1,6 @@
from typing import Optional, List
from sqlalchemy import Integer, String, ForeignKey, UniqueConstraint, ForeignKeyConstraint
from decimal import Decimal
from sqlalchemy import Integer, String, ForeignKey, UniqueConstraint, ForeignKeyConstraint, Numeric
from sqlalchemy.orm import Mapped, mapped_column, relationship
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
@@ -50,6 +51,10 @@ class EquivalencyItem(Base, TenantScopedMixin, TimestampMixin):
external_field: Mapped[str] = mapped_column(String(100), nullable=False)
conversion_factor: Mapped[Optional[Decimal]] = mapped_column(
Numeric(13, 6), nullable=True
)
equivalency: Mapped["Equivalency"] = relationship(back_populates="items")
unit_of_measure: Mapped["UnitOfMeasure"] = relationship()

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

@@ -38,6 +38,7 @@ item_crud = TenantCRUDRoutes(
tags=["Equivalency Items"],
resource_name="EquivalencyItem",
enable_list=True,
enable_filters=True,
)
# Custom endpoint for creating items nested under equivalency

252
backend/api/v1/modules/a76/general_catalogs/equivalencies/service.py
View File

@@ -1,6 +1,5 @@
from typing import List, Optional, Tuple, Dict, Any
from sqlalchemy.orm import Session, joinedload
from sqlalchemy import select
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
@@ -20,27 +19,26 @@ class EquivalencyService:
) -> Tuple[List[Equivalency], int]:
query = db.query(Equivalency).filter(
Equivalency.tenant_id == tenant_id,
Equivalency.company_id == company_id
).options(joinedload(Equivalency.items))
Equivalency.company_id == company_id,
)
if filters:
if 'fraccion_mex' in filters:
query = query.filter(Equivalency.identifier.ilike(f"%{filters['fraccion_mex']}%"))
if 'description' in filters:
query = query.filter(Equivalency.description.ilike(f"%{filters['description']}%"))
# Soporta filtros de UI por nombre de parámetro.
if "from_unit_code" in filters:
query = query.filter(
Equivalency.identifier.ilike(f"%{filters['from_unit_code']}%")
)
if "identifier" in filters:
query = query.filter(
Equivalency.identifier.ilike(f"%{filters['identifier']}%")
)
if "description" in filters:
query = query.filter(
Equivalency.description.ilike(f"%{filters['description']}%")
)
total = query.count()
items = query.offset(skip).limit(limit).all()
# Map internal fields to DTO fields
for item in items:
item.fraccion_mex = item.identifier
# Try to find the first item to get fraccion_us
if item.items:
item.fraccion_us = item.items[0].external_field
else:
item.fraccion_us = ""
return items, total
@staticmethod
@@ -54,15 +52,8 @@ class EquivalencyService:
Equivalency.id == id,
Equivalency.tenant_id == tenant_id,
Equivalency.company_id == company_id
).options(joinedload(Equivalency.items)).first()
if item:
item.fraccion_mex = item.identifier
if item.items:
item.fraccion_us = item.items[0].external_field
else:
item.fraccion_us = ""
).first()
return item
@staticmethod
@@ -73,53 +64,27 @@ class EquivalencyService:
company_id: int
) -> Equivalency:
try:
# Create Parent
db_obj = Equivalency(
identifier=data.fraccion_mex,
identifier=data.identifier,
description=data.description,
tenant_id=tenant_id,
company_id=company_id
company_id=company_id,
)
db.add(db_obj)
db.flush() # Flush to get ID
# Create Child Item (mapping fraccion_mex -> original_field, fraccion_us -> external_field)
item = EquivalencyItem(
equivalency_id=db_obj.id,
original_field=data.fraccion_mex, # Must exist in units_of_measure
external_field=data.fraccion_us,
tenant_id=tenant_id,
company_id=company_id
)
db.add(item)
db.commit()
db.refresh(db_obj)
# Map for response
db_obj.fraccion_mex = db_obj.identifier
db_obj.fraccion_us = item.external_field
return db_obj
except IntegrityError as e:
db.rollback()
error_msg = str(e.orig) if hasattr(e, 'orig') else str(e)
print(f"IntegrityError in create: {error_msg}")
if "units_of_measure" in error_msg:
raise HTTPException(
status_code=400,
detail=f"La Fracción MX '{data.fraccion_mex}' no es válida. Debe existir en el catálogo de Unidades de Medida."
)
error_msg = str(e.orig) if hasattr(e, "orig") else str(e)
if "uq_equivalency_identifier" in error_msg:
raise HTTPException(
status_code=400,
detail=f"Ya existe una equivalencia para la Fracción MX '{data.fraccion_mex}'."
)
raise HTTPException(status_code=400, detail=f"Error al guardar: {error_msg}")
status_code=400,
detail=f"Ya existe un registro para el origen '{data.identifier}'.",
) from e
raise HTTPException(status_code=400, detail=f"Error al guardar: {error_msg}") from e
except Exception as e:
db.rollback()
print(f"Error in create: {str(e)}")
raise e
@staticmethod
@@ -135,58 +100,19 @@ class EquivalencyService:
return None
try:
if data.fraccion_mex:
db_obj.identifier = data.fraccion_mex
if data.description:
db_obj.description = data.description
# Update Item
item = None
if db_obj.items:
item = db_obj.items[0]
if item:
if data.fraccion_us:
item.external_field = data.fraccion_us
if data.fraccion_mex:
item.original_field = data.fraccion_mex
else:
# Create if missing
if data.fraccion_us or data.fraccion_mex:
item = EquivalencyItem(
equivalency_id=db_obj.id,
original_field=data.fraccion_mex or db_obj.identifier,
external_field=data.fraccion_us or "",
tenant_id=tenant_id,
company_id=company_id
)
db.add(item)
update_dict = data.model_dump(exclude_unset=True)
for key, value in update_dict.items():
setattr(db_obj, key, value)
db.commit()
db.refresh(db_obj)
# Map for response
db_obj.fraccion_mex = db_obj.identifier
if db_obj.items:
db_obj.fraccion_us = db_obj.items[0].external_field
else:
db_obj.fraccion_us = ""
return db_obj
except IntegrityError as e:
db.rollback()
error_msg = str(e.orig) if hasattr(e, 'orig') else str(e)
print(f"IntegrityError in update: {error_msg}")
if "units_of_measure" in error_msg:
raise HTTPException(
status_code=400,
detail=f"La Fracción MX '{data.fraccion_mex or db_obj.identifier}' no es válida. Debe existir en el catálogo de Unidades de Medida."
)
raise HTTPException(status_code=400, detail=f"Error al actualizar: {error_msg}")
error_msg = str(e.orig) if hasattr(e, "orig") else str(e)
raise HTTPException(status_code=400, detail=f"Error al actualizar: {error_msg}") from e
except Exception as e:
db.rollback()
print(f"Error in update: {str(e)}")
raise e
@staticmethod
@@ -216,8 +142,21 @@ class EquivalencyItemService:
) -> Tuple[List[EquivalencyItem], int]:
query = db.query(EquivalencyItem).filter(
EquivalencyItem.tenant_id == tenant_id,
EquivalencyItem.company_id == company_id
EquivalencyItem.company_id == company_id,
)
if filters:
if "from_unit_code" in filters:
query = query.filter(
EquivalencyItem.original_field.ilike(
f"%{filters['from_unit_code']}%"
)
)
if "to_unit_code" in filters:
query = query.filter(
EquivalencyItem.external_field.ilike(f"%{filters['to_unit_code']}%")
)
total = query.count()
items = query.offset(skip).limit(limit).all()
return items, total
@@ -242,15 +181,51 @@ class EquivalencyItemService:
tenant_id: int,
company_id: int
) -> EquivalencyItem:
db_obj = EquivalencyItem(
**data.model_dump(),
tenant_id=tenant_id,
company_id=company_id
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
try:
# Equivalency padre agrupado por `from_unit_code` (original_field)
parent = (
db.query(Equivalency)
.filter(
Equivalency.tenant_id == tenant_id,
Equivalency.company_id == company_id,
Equivalency.identifier == data.original_field,
)
.first()
)
if not parent:
parent = Equivalency(
identifier=data.original_field,
description=None,
tenant_id=tenant_id,
company_id=company_id,
)
db.add(parent)
db.flush() # obtener id sin commit
db_obj = EquivalencyItem(
equivalency_id=parent.id,
original_field=data.original_field,
external_field=data.external_field,
conversion_factor=data.conversion_factor,
tenant_id=tenant_id,
company_id=company_id,
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
except IntegrityError as e:
db.rollback()
error_msg = str(e.orig) if hasattr(e, "orig") else str(e)
if "units_of_measure" in error_msg:
raise HTTPException(
status_code=400,
detail=(
"Verifica que las unidades existan en el catálogo de U.M."
),
) from e
raise HTTPException(status_code=400, detail=f"Error al guardar: {error_msg}") from e
@staticmethod
def create_nested(
@@ -264,6 +239,7 @@ class EquivalencyItemService:
equivalency_id=equivalency_id,
original_field=data.original_field,
external_field=data.external_field,
conversion_factor=data.conversion_factor,
tenant_id=tenant_id,
company_id=company_id
)
@@ -283,13 +259,47 @@ class EquivalencyItemService:
db_obj = EquivalencyItemService.get_by_id(db, id, tenant_id, company_id)
if not db_obj:
return None
for key, value in data.model_dump(exclude_unset=True).items():
setattr(db_obj, key, value)
db.commit()
db.refresh(db_obj)
return db_obj
update_dict = data.model_dump(exclude_unset=True)
try:
new_original = update_dict.get("original_field")
if new_original and new_original != db_obj.original_field:
# Re-asignar al padre del nuevo `from_unit_code`
parent = (
db.query(Equivalency)
.filter(
Equivalency.tenant_id == tenant_id,
Equivalency.company_id == company_id,
Equivalency.identifier == new_original,
)
.first()
)
if not parent:
parent = Equivalency(
identifier=new_original,
description=None,
tenant_id=tenant_id,
company_id=company_id,
)
db.add(parent)
db.flush()
db_obj.equivalency_id = parent.id
db_obj.original_field = new_original
# Aplicar resto de cambios
for key, value in update_dict.items():
if key == "original_field":
continue # ya aplicado arriba (y con posible re-parenting)
setattr(db_obj, key, value)
db.commit()
db.refresh(db_obj)
return db_obj
except IntegrityError as e:
db.rollback()
error_msg = str(e.orig) if hasattr(e, "orig") else str(e)
raise HTTPException(status_code=400, detail=f"Error al actualizar: {error_msg}") from e
@staticmethod
def delete(