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

Merge branch 'features/Creacion-formularios_catalogos_generales' into development

Browse Source
This commit is contained in:
AlexeerCT
2025-12-24 15:05:09 -06:00
parent 761271a664 b011ae32a4
commit 5ceaba50e5
279 changed files with 23252 additions and 3644 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,5 +14,6 @@ class ClassificationConcept(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
classification: Mapped[str] = mapped_column(
String(30), nullable=False) # CLASIFICACION

19
backend/api/v1/modules/a76/general_catalogs/classification_concepts/service.py
View File

@@ -1,10 +1,15 @@
from typing import List, Optional, Tuple, Dict, Any
from sqlalchemy.orm import Session
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
import logging
from .models import ClassificationConcept
from .dto import ClassificationConceptCreate, ClassificationConceptUpdate
logger = logging.getLogger(__name__)
class ClassificationConceptService:
@staticmethod
@@ -76,6 +81,14 @@ class ClassificationConceptService:
if not db_obj:
return False
db.delete(db_obj)
db.commit()
return True
try:
db.delete(db_obj)
db.commit()
return True
except IntegrityError as e:
db.rollback()
logger.error(f"Error de integridad al eliminar clasificación de concepto {id}: {str(e)}")
raise HTTPException(
status_code=400,
detail="No se puede eliminar esta clasificación porque tiene registros relacionados (pedimentos, facturas, etc.). Primero debe eliminar o reasignar esos registros."
)

176
backend/api/v1/modules/a76/general_catalogs/company/routes.py
View File

@@ -2,7 +2,7 @@
Rutas para gestión de empresa
"""
from typing import List
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
@@ -17,7 +17,78 @@ from .service import CompanyService
# Main router that includes base CRUD
router = APIRouter(prefix="/company")
# Custom endpoints
@router.post(
"", # Se suma al prefix, queda POST /api/v1/a76/company
response_model=CompanyResponseDTO,
status_code=status.HTTP_201_CREATED,
summary="Create a new company",
)
async def create_company(
data: CompanyCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
service = CompanyService(db)
return service.create_company_manually(data, tenant_id=tenant_id)
@router.get(
"", # GET /api/v1/a76/company with pagination
response_model=dict,
summary="Get companies with pagination",
)
async def list_companies(
page: int = 1,
page_size: int = 50,
name: Optional[str] = None,
rfc: Optional[str] = None,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""Get paginated list of companies for current tenant with optional filters"""
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
skip = (page - 1) * page_size
filters = {}
if name:
filters["name"] = name
if rfc:
filters["rfc"] = rfc
service = CompanyService(db)
items, total = service.get_all(
db,
tenant_id,
company_id=0, # Not used for companies
skip=skip,
limit=page_size,
filters=filters if filters else None
)
total_pages = (total + page_size - 1) // page_size
return {
"items": [CompanyResponseDTO.model_validate(item) for item in items],
"total": total,
"page": page,
"page_size": page_size,
"pages": total_pages,
}
@router.get(
"/my-companies",
response_model=List[CompanyResponseDTO],
@@ -166,17 +237,90 @@ async def get_program_info(
"prosec": company.prosec,
"prosec_authorization": company.prosec_authorization,
}
# Base CRUD routes using TenantCRUDRoutes
base_router = TenantCRUDRoutes(
service=CompanyService,
create_schema=CompanyCreateDTO,
update_schema=CompanyUpdateDTO,
response_schema=CompanyResponseDTO,
prefix="",
tags=[],
id_name="id",
enable_list=True,
enable_filters=True,
).router
router.include_router(base_router)
@router.get(
"/{company_id}",
response_model=CompanyResponseDTO,
summary="Get company by ID",
)
async def get_company(
company_id: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""Get a specific company by ID"""
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
company = CompanyService.get_by_id(db, company_id, tenant_id, 0)
if not company:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Company not found",
)
return CompanyResponseDTO.model_validate(company)
@router.put(
"/{company_id}",
response_model=CompanyResponseDTO,
summary="Update company",
)
async def update_company(
company_id: int,
data: CompanyUpdateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""Update a company"""
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
updated_company = CompanyService.update(
db, company_id, tenant_id, 0, data
)
if not updated_company:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Company not found",
)
return CompanyResponseDTO.model_validate(updated_company)
@router.delete(
"/{company_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete company",
)
async def delete_company(
company_id: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""Delete a company"""
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
success = CompanyService.delete(db, company_id, tenant_id, 0)
if not success:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Company not found",
)
return None

43
backend/api/v1/modules/a76/general_catalogs/company/service.py
View File

@@ -64,6 +64,7 @@ class CompanyService:
.first()
)
# ESTE ES EL MÉTODO VIEJO QUE CAUSABA PROBLEMAS (Lo dejamos por si acaso)
@staticmethod
def create(
db: Session,
@@ -71,7 +72,7 @@ class CompanyService:
tenant_id: int,
company_id: int,
) -> Company:
"""Create a new company"""
"""Create a new company (MÉTODO GENÉRICO - NO USAR PARA CREACIÓN MANUAL)"""
try:
db_company = Company(
**company_data.model_dump(exclude_unset=True),
@@ -136,10 +137,20 @@ class CompanyService:
db.delete(company)
db.commit()
return True
except IntegrityError as e:
db.rollback()
logger.error(f"IntegrityError deleting company {company_id}: {str(e)}")
# Check if it's a foreign key constraint
if "foreign key constraint" in str(e).lower():
raise HTTPException(
status_code=400,
detail="No se puede eliminar la empresa porque tiene registros relacionados (facturas, conceptos, etc.)"
)
raise HTTPException(status_code=400, detail="Error al eliminar la empresa")
except Exception as e:
db.rollback()
logger.error(f"Error deleting company {company_id}: {str(e)}")
raise HTTPException(status_code=500, detail="Error deleting company")
raise HTTPException(status_code=500, detail="Error al eliminar la empresa")
# Custom methods
def get_companies_by_tenant(self, tenant_id: int) -> List[Company]:
@@ -159,3 +170,31 @@ class CompanyService:
.first()
is not None
)
def create_company_manually(self, data: CompanyCreateDTO, tenant_id: int) -> Company:
try:
# 1. Preparar datos
obj_data = data.model_dump(exclude_unset=True)
# 2. Crear objeto SQLAlchemy
db_obj = Company(**obj_data, tenant_id=tenant_id)
# 3. Guardar
self.db.add(db_obj)
self.db.commit()
self.db.refresh(db_obj)
return db_obj
except IntegrityError as e:
self.db.rollback()
logger.error(f"IntegrityError creating company manually: {str(e)}")
raise HTTPException(
status_code=400,
detail="Error de integridad: Es posible que esta empresa ya exista.",
)
except Exception as e:
self.db.rollback()
logger.error(f"Error creating company manually: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error creando empresa: {str(e)}")

4
backend/api/v1/modules/a76/general_catalogs/concepts/dto.py
View File

@@ -4,6 +4,7 @@ from pydantic import BaseModel, Field, ConfigDict
class ConceptBase(BaseModel):
code: str = Field(..., max_length=15, description="Concept Code (CLAVE)")
company_id: int = Field(..., description="Company ID")
description: Optional[str] = Field(
None, max_length=120, description="Description")
description_en: Optional[str] = Field(
@@ -18,7 +19,6 @@ class ConceptBase(BaseModel):
section: Optional[int] = Field(None, description="Section")
classification: Optional[str] = Field(
None, max_length=30, description="Classification")
company_id: int = Field(..., description="Company ID")
class ConceptCreate(ConceptBase):
@@ -32,7 +32,7 @@ class ConceptUpdate(BaseModel):
detailed_description: Optional[str] = Field(None, max_length=1000)
priority: Optional[int] = None
priority_ame: Optional[int] = None
first_total: Optional[bool] = None
first_total: Optional[bool] = None
type: Optional[str] = Field(None, max_length=9)
is_printed: Optional[bool] = None
section: Optional[int] = None

14
backend/api/v1/modules/a76/general_catalogs/concepts/models.py
View File

@@ -18,25 +18,39 @@ class Concept(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
company_id: Mapped[int] = mapped_column(
Integer, nullable=False) # IDEMPRESA
code: Mapped[str] = mapped_column(String(15), nullable=False) # CLAVE
description: Mapped[Optional[str]] = mapped_column(
String(120), nullable=True) # DESCRIPCION
description_en: Mapped[Optional[str]] = mapped_column(
String(120), nullable=True) # DESCRIPCIONINGLES
detailed_description: Mapped[Optional[str]] = mapped_column(
String(1000), nullable=True) # DESCRIPCIONDETALLADA
priority: Mapped[Optional[int]] = mapped_column(
Integer, nullable=True) # PRIORIDAD
priority_ame: Mapped[Optional[int]] = mapped_column(
Integer, nullable=True) # PRIORIDADAME
first_total: Mapped[Optional[bool]] = mapped_column(
Boolean, nullable=True) # PRIMERTOTAL
type: Mapped[Optional[str]] = mapped_column(
String(9), nullable=True) # TIPO
is_printed: Mapped[Optional[bool]] = mapped_column(
Boolean, nullable=True) # SEIMPRIME
section: Mapped[Optional[int]] = mapped_column(
Integer, nullable=True) # SECCION
classification: Mapped[Optional[str]] = mapped_column(String(30), ForeignKey(
"a76.classification_concepts.classification"), nullable=True) # CLASIFICACION

26
backend/api/v1/modules/a76/general_catalogs/concepts/service.py
View File

@@ -1,10 +1,16 @@
from typing import List, Optional, Tuple, Dict, Any
import logging
from sqlalchemy.orm import Session
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
from .models import Concept
from .dto import ConceptCreate, ConceptUpdate
logger = logging.getLogger(__name__)
class ConceptService:
@staticmethod
@@ -74,6 +80,20 @@ class ConceptService:
if not db_obj:
return False
db.delete(db_obj)
db.commit()
return True
try:
db.delete(db_obj)
db.commit()
return True
except IntegrityError as e:
db.rollback()
logger.error(f"IntegrityError deleting concept {id}: {str(e)}")
if "foreign key constraint" in str(e).lower():
raise HTTPException(
status_code=400,
detail="No se puede eliminar el concepto porque tiene registros relacionados"
)
raise HTTPException(status_code=400, detail="Error al eliminar el concepto")
except Exception as e:
db.rollback()
logger.error(f"Error deleting concept {id}: {str(e)}")
raise HTTPException(status_code=500, detail="Error al eliminar el concepto")

3
backend/api/v1/modules/a76/general_catalogs/customs_broker_concepts/dto.py
View File

@@ -4,8 +4,7 @@ from pydantic import BaseModel, Field, ConfigDict
class CustomsBrokerConceptBase(BaseModel):
broker_key: str = Field(..., max_length=5,
description="Customs Broker Key (CLAVEAA)")
broker_key: str = Field(..., max_length=5, description="Customs Broker Key (CLAVEAA)")
concept: str = Field(..., max_length=15, description="Concept")
amount: Optional[Decimal] = Field(None, description="Amount")
priority: Optional[int] = Field(None, description="Priority")

8
backend/api/v1/modules/a76/general_catalogs/customs_broker_concepts/models.py
View File

@@ -9,17 +9,23 @@ from core.database import Base
class CustomsBrokerConcept(Base, TenantScopedMixin, TimestampMixin):
__tablename__ = "customs_broker_concepts"
__table_args__ = (
UniqueConstraint("broker_key", "concept", name="uq_broker_concept"),
UniqueConstraint("broker_key", "concept", "company_id", name="uq_broker_concept"),
{"schema": "a76"}
)
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
company_id: Mapped[int] = mapped_column(Integer, nullable=False)
broker_key: Mapped[str] = mapped_column(
String(5), nullable=False) # CLAVEAA
concept: Mapped[str] = mapped_column(
String(15), nullable=False) # CONCEPTO
amount: Mapped[Optional[Decimal]] = mapped_column(
Numeric(11, 2), nullable=True) # IMPORTE
priority: Mapped[Optional[int]] = mapped_column(
Integer, nullable=True) # PRIORIDAD

2
backend/api/v1/modules/a76/general_catalogs/customs_broker_concepts/routes.py
View File

@@ -7,7 +7,7 @@ router = TenantCRUDRoutes(
create_schema=CustomsBrokerConceptCreate,
update_schema=CustomsBrokerConceptUpdate,
response_schema=CustomsBrokerConceptResponse,
prefix="/customs-broker-concepts",
prefix="/customs-broker-concepts",
tags=["a76.general_catalogs.customs_broker_concepts"],
resource_name="Customs Broker Concept",
enable_list=True,

26
backend/api/v1/modules/a76/general_catalogs/customs_broker_concepts/service.py
View File

@@ -1,10 +1,16 @@
from typing import List, Optional, Tuple, Dict, Any
import logging
from sqlalchemy.orm import Session
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
from .models import CustomsBrokerConcept
from .dto import CustomsBrokerConceptCreate, CustomsBrokerConceptUpdate
logger = logging.getLogger(__name__)
class CustomsBrokerConceptService:
@staticmethod
@@ -76,6 +82,20 @@ class CustomsBrokerConceptService:
if not db_obj:
return False
db.delete(db_obj)
db.commit()
return True
try:
db.delete(db_obj)
db.commit()
return True
except IntegrityError as e:
db.rollback()
logger.error(f"IntegrityError deleting customs broker concept {id}: {str(e)}")
if "foreign key constraint" in str(e).lower():
raise HTTPException(
status_code=400,
detail="No se puede eliminar el concepto porque tiene registros relacionados"
)
raise HTTPException(status_code=400, detail="Error al eliminar el concepto")
except Exception as e:
db.rollback()
logger.error(f"Error deleting customs broker concept {id}: {str(e)}")
raise HTTPException(status_code=500, detail="Error al eliminar el concepto")

4
backend/api/v1/modules/a76/general_catalogs/doda/dto.py
View File

@@ -340,7 +340,7 @@ class DodaUpdateDTO(BaseModel):
class DodaResponseDTO(BaseModel):
"""DTO para responder con datos de un DODA"""
sys_id: int
id: int
integration_number: Optional[str] = None
doda_date: Optional[int] = None
doda_time: Optional[int] = None
@@ -383,7 +383,7 @@ class DodaResponseDTO(BaseModel):
class DodaDetailResponseDTO(BaseModel):
"""DTO detallado para responder con todos los datos de un DODA"""
sys_id: int
id: int
integration_number: Optional[str] = None
doda_date: Optional[int] = None
doda_time: Optional[int] = None

8
backend/api/v1/modules/a76/general_catalogs/doda/models.py
View File

@@ -151,19 +151,19 @@ class DodaContainerSeal(Base, TenantScopedMixin, TimestampMixin):
{"schema": "a76"},
)
# Primary key
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
# Foreign key and line info
container_id: Mapped[int] = mapped_column(Integer, nullable=False)
doda_id: Mapped[int] = mapped_column(Integer, nullable=False)
seal_line: Mapped[int] = mapped_column(Integer, nullable=False)
# Seal information
seal_value: Mapped[Optional[str]] = mapped_column(String(21))
# Relationships
container: Mapped["DodaContainer"] = relationship(
"DodaContainer", back_populates="seals_detail"
)

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

@@ -131,14 +131,21 @@ class DodaService:
db: Session, id: int, tenant_id: int, company_id: int
) -> bool:
"""Delete a DODA"""
try:
db_doda = DodaService.get_by_id(db, id, tenant_id, company_id)
if not db_doda:
return False
db_doda = DodaService.get_by_id(db, id, tenant_id, company_id)
if not db_doda:
return False
try:
db.delete(db_doda)
db.commit()
return True
except IntegrityError as e:
db.rollback()
logger.error(f"Error de integridad al eliminar DODA {id}: {str(e)}")
raise HTTPException(
status_code=400,
detail="No se puede eliminar este DODA porque tiene registros relacionados. Primero debe eliminar o reasignar esos registros."
)
except Exception as e:
db.rollback()
logger.error(f"Error deleting DODA: {str(e)}")

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

@@ -68,7 +68,7 @@ class ElectronicNoticeUpdateDTO(BaseModel):
class ElectronicNoticeResponseDTO(BaseModel):
"""DTO para responder con datos de un aviso electrónico"""
sys_id: int
id: int
notice_number: Optional[str] = None
year: Optional[str] = None
patent: Optional[str] = None

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

@@ -30,22 +30,27 @@ class EquivalencyItemResponse(EquivalencyItemBase):
class EquivalencyBase(BaseModel):
identifier: str = Field(..., max_length=10, description="Identifier")
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")
class EquivalencyCreate(EquivalencyBase):
items: Optional[List[EquivalencyItemCreate]] = []
pass
class EquivalencyUpdate(BaseModel):
identifier: Optional[str] = Field(None, max_length=10)
fraccion_mex: Optional[str] = Field(None, max_length=10)
fraccion_us: Optional[str] = Field(None, max_length=100)
description: Optional[str] = Field(None, max_length=200)
class EquivalencyResponse(EquivalencyBase):
id: int
items: List[EquivalencyItemResponse] = []
tenant_id: int
company_id: int
model_config = ConfigDict(from_attributes=True)
model_config = ConfigDict(from_attributes=True)

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

@@ -16,7 +16,9 @@ class Equivalency(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
identifier: Mapped[str] = mapped_column(String(10), nullable=False)
description: Mapped[Optional[str]] = mapped_column(
String(200), nullable=True)
@@ -39,11 +41,15 @@ class EquivalencyItem(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
equivalency_id: Mapped[int] = mapped_column(
Integer, ForeignKey("a76.equivalencies.id"), nullable=False)
original_field: Mapped[str] = mapped_column(
String(100), nullable=False) # Relation to Unit of Measure
external_field: Mapped[str] = mapped_column(String(100), nullable=False)
equivalency: Mapped["Equivalency"] = relationship(back_populates="items")
unit_of_measure: Mapped["UnitOfMeasure"] = relationship()

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

@@ -1,6 +1,7 @@
from typing import List, Optional, Tuple, Dict, Any
from sqlalchemy.orm import Session
from sqlalchemy.orm import Session, joinedload
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
from .models import Equivalency, EquivalencyItem
@@ -20,14 +21,26 @@ class EquivalencyService:
query = db.query(Equivalency).filter(
Equivalency.tenant_id == tenant_id,
Equivalency.company_id == company_id
)
).options(joinedload(Equivalency.items))
if filters:
# Add filters here if needed
pass
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']}%"))
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
@@ -37,11 +50,20 @@ class EquivalencyService:
tenant_id: int,
company_id: int
) -> Optional[Equivalency]:
return db.query(Equivalency).filter(
item = db.query(Equivalency).filter(
Equivalency.id == id,
Equivalency.tenant_id == tenant_id,
Equivalency.company_id == company_id
).first()
).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 = ""
return item
@staticmethod
def create(
@@ -50,50 +72,122 @@ class EquivalencyService:
tenant_id: int,
company_id: int
) -> Equivalency:
db_obj = Equivalency(
identifier=data.identifier,
description=data.description,
tenant_id=tenant_id,
company_id=company_id
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
try:
# Create Parent
db_obj = Equivalency(
identifier=data.fraccion_mex,
description=data.description,
tenant_id=tenant_id,
company_id=company_id
)
db.add(db_obj)
db.flush() # Flush to get ID
if data.items:
for item_data in data.items:
item = EquivalencyItem(
**item_data.model_dump(),
equivalency_id=db_obj.id,
tenant_id=tenant_id,
company_id=company_id
)
db.add(item)
# 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)
return 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."
)
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}")
except Exception as e:
db.rollback()
print(f"Error in create: {str(e)}")
raise e
@staticmethod
def update(
db: Session,
id: int,
data: EquivalencyUpdate,
tenant_id: int,
data: EquivalencyUpdate,
company_id: int
) -> Optional[Equivalency]:
db_obj = EquivalencyService.get_by_id(db, id, tenant_id, company_id)
if not db_obj:
return None
update_dict = data.model_dump(exclude_unset=True)
for key, value in update_dict.items():
if key != 'items': # Handle items separately if needed, or ignore for now as per original code
setattr(db_obj, key, value)
try:
if data.fraccion_mex:
db_obj.identifier = data.fraccion_mex
if data.description:
db_obj.description = data.description
db.commit()
db.refresh(db_obj)
return db_obj
# 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)
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}")
except Exception as e:
db.rollback()
print(f"Error in update: {str(e)}")
raise e
@staticmethod
def delete(
@@ -110,7 +204,6 @@ class EquivalencyService:
db.commit()
return True
class EquivalencyItemService:
@staticmethod
def get_all(
@@ -125,11 +218,6 @@ class EquivalencyItemService:
EquivalencyItem.tenant_id == tenant_id,
EquivalencyItem.company_id == company_id
)
if filters and "equivalency_id" in filters:
query = query.filter(
EquivalencyItem.equivalency_id == filters["equivalency_id"])
total = query.count()
items = query.offset(skip).limit(limit).all()
return items, total
@@ -154,13 +242,6 @@ class EquivalencyItemService:
tenant_id: int,
company_id: int
) -> EquivalencyItem:
# Note: equivalency_id should be in data or handled by the caller if it's a nested route
# But TenantCRUDRoutes for child resources might pass it in filters or we need to handle it.
# For now, assuming it's in data or we don't use child resource feature yet.
# If using child resource, the parent_id is usually passed in the path.
# But TenantCRUDRoutes passes the body.
db_obj = EquivalencyItem(
**data.model_dump(),
tenant_id=tenant_id,
@@ -180,8 +261,9 @@ class EquivalencyItemService:
company_id: int
) -> EquivalencyItem:
db_obj = EquivalencyItem(
**data.model_dump(),
equivalency_id=equivalency_id,
original_field=data.original_field,
external_field=data.external_field,
tenant_id=tenant_id,
company_id=company_id
)
@@ -194,19 +276,17 @@ class EquivalencyItemService:
def update(
db: Session,
id: int,
data: EquivalencyItemUpdate,
tenant_id: int,
data: EquivalencyItemUpdate,
company_id: int
) -> Optional[EquivalencyItem]:
db_obj = EquivalencyItemService.get_by_id(
db, id, tenant_id, company_id)
db_obj = EquivalencyItemService.get_by_id(db, id, tenant_id, company_id)
if not db_obj:
return None
update_dict = data.model_dump(exclude_unset=True)
for key, value in update_dict.items():
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
@@ -218,11 +298,9 @@ class EquivalencyItemService:
tenant_id: int,
company_id: int
) -> bool:
db_obj = EquivalencyItemService.get_by_id(
db, id, tenant_id, company_id)
db_obj = EquivalencyItemService.get_by_id(db, id, tenant_id, company_id)
if not db_obj:
return False
db.delete(db_obj)
db.commit()
return True

8
backend/api/v1/modules/a76/general_catalogs/error_catalogs/models.py
View File

@@ -24,7 +24,8 @@ class ErrorClassification(Base, TenantScopedMixin, TimestampMixin):
# Primary key
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
Integer, primary_key=True, autoincrement=True
)
# Classification code (unique)
code: Mapped[str] = mapped_column(String(100), nullable=False, unique=True)
@@ -60,7 +61,8 @@ class ErrorCatalog(Base, TenantScopedMixin, TimestampMixin):
# Primary key
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
Integer, primary_key=True, autoincrement=True
)
# Error code (unique)
code: Mapped[str] = mapped_column(String(15), nullable=False, unique=True)
@@ -77,4 +79,4 @@ class ErrorCatalog(Base, TenantScopedMixin, TimestampMixin):
)
def __repr__(self):
return f"<ErrorCatalog(id={self.id}, code={self.code}, description={self.description}, classification_id={self.classification_id})>"
return f"<ErrorCatalog(id={self.id}, code={self.code}, description={self.description}, classification_id={self.classification_id})>"

3
backend/api/v1/modules/a76/general_catalogs/exchange_rate/models.py
View File

@@ -30,6 +30,9 @@ class ExchangeRate(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(Integer, primary_key=True)
date: Mapped[datetime] = mapped_column(DateTime)
value: Mapped[Optional[Decimal]] = mapped_column(DECIMAL(13, 6))
local_currency: Mapped[Optional[str]] = mapped_column(String(7))
foreign_currency: Mapped[Optional[str]] = mapped_column(String(7))

17
backend/api/v1/modules/a76/general_catalogs/identifiers/dto.py
View File

@@ -1,9 +1,6 @@
from typing import Optional
from pydantic import BaseModel, Field, ConfigDict
# Identifier DTOs
class IdentifierBase(BaseModel):
code: str = Field(..., max_length=2, description="Identifier Code (CLAVE)")
description: Optional[str] = Field(
@@ -11,28 +8,23 @@ class IdentifierBase(BaseModel):
level: Optional[str] = Field(None, max_length=1, description="Level")
complement: Optional[str] = Field(
None, max_length=5000, description="Complement")
company_id: int = Field(..., description="Company ID")
class IdentifierCreate(IdentifierBase):
pass
class IdentifierUpdate(BaseModel):
code: Optional[str] = Field(None, max_length=2)
description: Optional[str] = Field(None, max_length=1000)
level: Optional[str] = Field(None, max_length=1)
complement: Optional[str] = Field(None, max_length=5000)
class IdentifierResponse(IdentifierBase):
id: int
tenant_id: int
company_id: int
model_config = ConfigDict(from_attributes=True)
# Identifier Detail DTOs
class IdentifierDetailBase(BaseModel):
invoice_consecutive: Optional[int] = Field(
@@ -47,13 +39,10 @@ class IdentifierDetailBase(BaseModel):
None, max_length=51, description="Complement 2")
complement3: Optional[str] = Field(
None, max_length=50, description="Complement 3")
company_id: int = Field(..., description="Company ID")
class IdentifierDetailCreate(IdentifierDetailBase):
pass
class IdentifierDetailUpdate(BaseModel):
invoice_consecutive: Optional[int] = None
part_line: Optional[int] = None
@@ -63,9 +52,9 @@ class IdentifierDetailUpdate(BaseModel):
complement2: Optional[str] = Field(None, max_length=51)
complement3: Optional[str] = Field(None, max_length=50)
class IdentifierDetailResponse(IdentifierDetailBase):
id: int
tenant_id: int
company_id: int
model_config = ConfigDict(from_attributes=True)
model_config = ConfigDict(from_attributes=True)

4
backend/api/v1/modules/a76/general_catalogs/identifiers/models.py
View File

@@ -14,11 +14,15 @@ class Identifier(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
code: Mapped[str] = mapped_column(String(2), nullable=False) # CLAVE
description: Mapped[Optional[str]] = mapped_column(
String(1000), nullable=True) # DESCRIPCION
level: Mapped[Optional[str]] = mapped_column(
String(1), nullable=True) # NIVEL
complement: Mapped[Optional[str]] = mapped_column(
String(5000), nullable=True) # COMPLEMENTO

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

@@ -68,8 +68,8 @@ class IdentifierService:
def update(
db: Session,
id: int,
data: IdentifierUpdate,
tenant_id: int,
data: IdentifierUpdate,
company_id: int
) -> Optional[Identifier]:
db_obj = IdentifierService.get_by_id(db, id, tenant_id, company_id)

6
backend/api/v1/modules/a76/general_catalogs/inpc/dto.py
View File

@@ -2,7 +2,6 @@ from typing import Optional
from decimal import Decimal
from pydantic import BaseModel, Field, ConfigDict
class INPCBase(BaseModel):
year: str = Field(..., max_length=4, description="Year (YYYY)")
month: str = Field(..., max_length=2, description="Month (MM)")
@@ -12,7 +11,6 @@ class INPCBase(BaseModel):
class INPCCreate(INPCBase):
pass
class INPCUpdate(BaseModel):
year: Optional[str] = Field(None, max_length=4)
month: Optional[str] = Field(None, max_length=2)
@@ -21,5 +19,7 @@ class INPCUpdate(BaseModel):
class INPCResponse(INPCBase):
id: int
tenant_id: int
company_id: int
model_config = ConfigDict(from_attributes=True)
model_config = ConfigDict(from_attributes=True)

3
backend/api/v1/modules/a76/general_catalogs/inpc/models.py
View File

@@ -16,7 +16,10 @@ class INPC(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
year: Mapped[str] = mapped_column(String(4), nullable=False) # ANIO
month: Mapped[str] = mapped_column(String(2), nullable=False) # MES
value: Mapped[Optional[Decimal]] = mapped_column(
Numeric(19, 8), nullable=True) # VALOR

14
backend/api/v1/modules/a76/general_catalogs/inpc/routes.py
View File

@@ -1,20 +1,16 @@
from fastapi import APIRouter
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from .models import INPC
from .dto import INPCCreate, INPCResponse, INPCUpdate
from .service import INPCService
router = APIRouter(prefix="/inpc", tags=["a76.general_catalogs.inpc"])
inpc_crud = TenantCRUDRoutes(
# Usamos TenantCRUDRoutes directamente
router = TenantCRUDRoutes(
service=INPCService,
create_schema=INPCCreate,
update_schema=INPCUpdate,
response_schema=INPCResponse,
prefix="",
tags=["INPC"],
prefix="/inpc",
tags=["a76.general_catalogs.inpc"],
resource_name="INPC",
enable_list=True,
)
router.include_router(inpc_crud.router)
).router

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

@@ -63,8 +63,8 @@ class INPCService:
def update(
db: Session,
id: int,
data: INPCUpdate,
tenant_id: int,
data: INPCUpdate,
company_id: int
) -> Optional[INPC]:
db_obj = INPCService.get_by_id(db, id, tenant_id, company_id)

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

@@ -21,3 +21,5 @@ class LegendResponse(LegendBase):
id: int
model_config = ConfigDict(from_attributes=True)
tenant_id : int
company_id : int

2
backend/api/v1/modules/a76/general_catalogs/legends/models.py
View File

@@ -15,6 +15,8 @@ class Legend(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
code: Mapped[int] = mapped_column(Integer, nullable=False) # CLAVELEY
description: Mapped[Optional[str]] = mapped_column(
String(2000), nullable=True) # DESCLEYENDA

12
backend/api/v1/modules/a76/general_catalogs/legends/routes.py
View File

@@ -4,17 +4,13 @@ from .models import Legend
from .dto import LegendCreate, LegendResponse, LegendUpdate
from .service import LegendService
router = APIRouter(prefix="/legends", tags=["a76.general_catalogs.legends"])
legend_crud = TenantCRUDRoutes(
router = TenantCRUDRoutes(
service=LegendService,
create_schema=LegendCreate,
update_schema=LegendUpdate,
response_schema=LegendResponse,
prefix="",
tags=["Legends"],
prefix="/legends",
tags=["a76.general_catalogs.legends"],
resource_name="Legend",
enable_list=True,
)
router.include_router(legend_crud.router)
).router

21
backend/api/v1/modules/a76/general_catalogs/legends/service.py
View File

@@ -1,10 +1,15 @@
from typing import List, Optional, Tuple, Dict, Any
from sqlalchemy.orm import Session
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
import logging
from .models import Legend
from .dto import LegendCreate, LegendUpdate
logger = logging.getLogger(__name__)
class LegendService:
@staticmethod
@@ -63,8 +68,8 @@ class LegendService:
def update(
db: Session,
id: int,
data: LegendUpdate,
tenant_id: int,
data: LegendUpdate,
company_id: int
) -> Optional[Legend]:
db_obj = LegendService.get_by_id(db, id, tenant_id, company_id)
@@ -90,6 +95,14 @@ class LegendService:
if not db_obj:
return False
db.delete(db_obj)
db.commit()
return True
try:
db.delete(db_obj)
db.commit()
return True
except IntegrityError as e:
db.rollback()
logger.error(f"Error de integridad al eliminar leyenda {id}: {str(e)}")
raise HTTPException(
status_code=400,
detail="No se puede eliminar esta leyenda porque tiene registros relacionados (pedimentos, facturas, etc.). Primero debe eliminar o reasignar esos registros."
)

0
backend/api/v1/modules/a76/general_catalogs/locations/__init__.py Normal file
View File

4
backend/api/v1/modules/a76/general_catalogs/multi_currency_types/dto.py
View File

@@ -27,5 +27,7 @@ class MultiCurrencyTypeUpdate(BaseModel):
class MultiCurrencyTypeResponse(MultiCurrencyTypeBase):
id: int
company_id: int
tenant_id: int
model_config = ConfigDict(from_attributes=True)

5
backend/api/v1/modules/a76/general_catalogs/multi_currency_types/models.py
View File

@@ -18,13 +18,18 @@ class MultiCurrencyType(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
currency_type_code: Mapped[str] = mapped_column(
String(3), ForeignKey("public.currency_types.code"), nullable=False)
country_key: Mapped[Optional[str]] = mapped_column(
String(3), ForeignKey("public.countries.m3_key"), nullable=True)
conversion_factor: Mapped[Optional[Decimal]] = mapped_column(
Numeric(13, 6), nullable=True)
publication_date: Mapped[int] = mapped_column(Integer, nullable=False)
currency_type: Mapped["CurrencyType"] = relationship()
country: Mapped[Optional["Country"]] = relationship()

2
backend/api/v1/modules/a76/general_catalogs/packages/models.py
View File

@@ -31,4 +31,4 @@ class Package(Base, TenantScopedMixin, TimestampMixin):
plurals: Mapped[Optional[str]] = mapped_column(String(4))
plural_in: Mapped[Optional[str]] = mapped_column(String(4))
code_ace: Mapped[Optional[str]] = mapped_column(String(4))
code_aamex: Mapped[Optional[str]] = mapped_column(String(9))
code_aamex: Mapped[Optional[str]] = mapped_column(String(9))

25
backend/api/v1/modules/a76/general_catalogs/packages/services.py
View File

@@ -3,11 +3,16 @@ Service layer for Packages (GBultos).
"""
from typing import Optional, Tuple, List, Dict, Any
import logging
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
from . import dto, models
logger = logging.getLogger(__name__)
class PackageService:
"""Service for Package CRUD operations with tenant support"""
@@ -109,6 +114,20 @@ class PackageService:
if not package:
return False
db.delete(package)
db.commit()
return True
try:
db.delete(package)
db.commit()
return True
except IntegrityError as e:
db.rollback()
logger.error(f"IntegrityError deleting package {package_id}: {str(e)}")
if "foreign key constraint" in str(e).lower():
raise HTTPException(
status_code=400,
detail="No se puede eliminar el bulto porque tiene registros relacionados"
)
raise HTTPException(status_code=400, detail="Error al eliminar el bulto")
except Exception as e:
db.rollback()
logger.error(f"Error deleting package {package_id}: {str(e)}")
raise HTTPException(status_code=500, detail="Error al eliminar el bulto")

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

@@ -63,8 +63,8 @@ class PortService:
def update(
db: Session,
id: int,
data: PortUpdate,
tenant_id: int,
data: PortUpdate,
company_id: int
) -> Optional[Port]:
db_obj = PortService.get_by_id(db, id, tenant_id, company_id)

29
backend/api/v1/modules/a76/general_catalogs/prevalidators/dto.py
View File

@@ -56,32 +56,3 @@ class PrevalidatorResponseDTO(BaseModel):
class Config:
from_attributes = True
class PrevalidatorUpdateDTO(BaseModel):
"""DTO para actualizar un prevalidador"""
customs_prevalidator: Optional[str] = Field(
None, max_length=20, description="Customs prevalidator"
)
patent_prevalidator: Optional[str] = Field(
None, max_length=20, description="Patent prevalidator"
)
description: Optional[str] = Field(
None, max_length=50, description="Description"
)
class Config:
from_attributes = True
class PrevalidatorResponseDTO(BaseModel):
"""DTO para responder con datos de un prevalidador"""
code: str
customs_prevalidator: Optional[str] = None
patent_prevalidator: Optional[str] = None
description: Optional[str] = None
class Config:
from_attributes = True

2
backend/api/v1/modules/a76/general_catalogs/prevalidators/models.py
View File

@@ -31,7 +31,9 @@ class Prevalidator(Base, TenantScopedMixin, TimestampMixin):
# Prevalidator information
customs_prevalidator: Mapped[Optional[str]] = mapped_column(String(20))
patent_prevalidator: Mapped[Optional[str]] = mapped_column(String(20))
description: Mapped[Optional[str]] = mapped_column(String(50))
def __repr__(self):

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

@@ -120,8 +120,8 @@ class PrevalidatorService:
def update(
db: Session,
prevalidator_id: int,
prevalidator_data: PrevalidatorUpdateDTO,
tenant_id: int,
prevalidator_data: PrevalidatorUpdateDTO,
company_id: int
) -> Optional[Prevalidator]:
"""Update a prevalidator"""

3
backend/api/v1/modules/a76/general_catalogs/signatures/dto.py
View File

@@ -11,6 +11,7 @@ class SignatureCreateDTO(BaseModel):
"""DTO para crear una firma"""
code: str = Field(..., max_length=10, description="Signature code")
signature: Optional[str] = Field(
None, max_length=1000, description="Signature")
photo_path: Optional[str] = Field(
@@ -39,6 +40,8 @@ class SignatureResponseDTO(BaseModel):
code: str
signature: Optional[str] = None
photo_path: Optional[str] = None
tenant_id: int
company_id: int
class Config:
from_attributes = True

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

@@ -32,6 +32,7 @@ class Signature(Base, TenantScopedMixin, TimestampMixin):
# Signature information
signature: Mapped[Optional[str]] = mapped_column(String(1000))
photo_path: Mapped[Optional[str]] = mapped_column(String(1000))
def __repr__(self):

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

@@ -5,6 +5,8 @@ Capa de servicio para lógica de negocio de firmas
import logging
from typing import Any, Dict, List, Optional, Tuple
from fastapi import HTTPException, status
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from . import dto, models
@@ -81,10 +83,18 @@ class SignatureService:
new_signature = models.Signature(
**signature_data.model_dump(), tenant_id=tenant_id, company_id=company_id
)
db.add(new_signature)
db.commit()
db.refresh(new_signature)
return new_signature
try:
db.add(new_signature)
db.commit()
db.refresh(new_signature)
return new_signature
except IntegrityError as exc:
db.rollback()
# Constraint names: signatures_code_unique (code, tenant_id, company_id)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Ya existe una firma con ese código para esta compañía",
) from exc
@staticmethod
def update(

18
backend/api/v1/modules/a76/general_catalogs/unit_conversions/dto.py
View File

@@ -2,27 +2,23 @@ from typing import Optional
from decimal import Decimal
from pydantic import BaseModel, Field, ConfigDict
class UnitConversionBase(BaseModel):
from_unit_code: str = Field(..., max_length=5,
description="Source Unit Code")
to_unit_code: str = Field(..., max_length=5,
description="Target Unit Code")
conversion_factor: Optional[Decimal] = Field(
None, description="Conversion Factor")
# 👇 OJO: Son códigos (Strings), no IDs
from_unit_code: str = Field(..., max_length=5, description="Source Unit Code")
to_unit_code: str = Field(..., max_length=5, description="Target Unit Code")
conversion_factor: Optional[Decimal] = Field(None, description="Conversion Factor")
class UnitConversionCreate(UnitConversionBase):
pass
class UnitConversionUpdate(BaseModel):
from_unit_code: Optional[str] = Field(None, max_length=5)
to_unit_code: Optional[str] = Field(None, max_length=5)
conversion_factor: Optional[Decimal] = None
class UnitConversionResponse(UnitConversionBase):
id: int
tenant_id: int
company_id: int
model_config = ConfigDict(from_attributes=True)
model_config = ConfigDict(from_attributes=True)

3
backend/api/v1/modules/a76/general_catalogs/unit_conversions/models.py
View File

@@ -27,8 +27,11 @@ class UnitConversion(Base, TenantScopedMixin, TimestampMixin):
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
from_unit_code: Mapped[str] = mapped_column(String(5), nullable=False)
to_unit_code: Mapped[str] = mapped_column(String(5), nullable=False)
conversion_factor: Mapped[Optional[Decimal]] = mapped_column(
Numeric(13, 6), nullable=True)

34
backend/api/v1/modules/a76/general_catalogs/unit_conversions/service.py
View File

@@ -1,6 +1,9 @@
from typing import List, Optional, Tuple, Dict, Any
from sqlalchemy.orm import Session
from fastapi import HTTPException, status
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from .models import UnitConversion
from .dto import UnitConversionCreate, UnitConversionUpdate
@@ -51,10 +54,18 @@ class UnitConversionService:
tenant_id=tenant_id,
company_id=company_id
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
try:
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
except IntegrityError as exc:
db.rollback()
# Puede ser FK de unidades o duplicado de (from_unit_code, to_unit_code, tenant, company)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Verifica que los códigos de unidad existan y que la conversión no esté duplicada",
) from exc
@staticmethod
def update(
@@ -72,9 +83,16 @@ class UnitConversionService:
for key, value in update_dict.items():
setattr(db_obj, key, value)
db.commit()
db.refresh(db_obj)
return db_obj
try:
db.commit()
db.refresh(db_obj)
return db_obj
except IntegrityError as exc:
db.rollback()
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Verifica que los códigos de unidad existan y que la conversión no esté duplicada",
) from exc
@staticmethod
def delete(

2
backend/api/v1/modules/a76/general_catalogs/units_of_measure/models.py
View File

@@ -181,4 +181,4 @@ class UnitOfMeasureGeneral(Base, TenantScopedMixin, TimestampMixin):
String(4), nullable=True) # CLAVEACE
customs_unit: Mapped[Optional["UnitOfMeasureCustoms"]] = relationship()
ace_unit: Mapped[Optional["UnitOfMeasureACE"]] = relationship()
ace_unit: Mapped[Optional["UnitOfMeasureACE"]] = relationship(overlaps="customs_unit")

19
backend/api/v1/modules/a76/general_catalogs/units_of_measure/service.py
View File

@@ -1,6 +1,11 @@
from typing import List, Optional, Tuple, Dict, Any, Type
from sqlalchemy import Sequence
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
import logging
logger = logging.getLogger(__name__)
from .models import (
UnitOfMeasureACE, UnitOfMeasureOMA, UnitOfMeasureAmerican, UnitOfMeasureCustoms,
@@ -102,9 +107,17 @@ class BaseService:
if not db_obj:
return False
db.delete(db_obj)
db.commit()
return True
try:
db.delete(db_obj)
db.commit()
return True
except IntegrityError as e:
db.rollback()
logger.error(f"Error de integridad al eliminar {cls.model.__name__} {id}: {str(e)}")
raise HTTPException(
status_code=400,
detail="No se puede eliminar esta unidad de medida porque tiene registros relacionados. Primero debe eliminar o reasignar esos registros."
)
class UnitOfMeasureACEService(BaseService):

259
backend/api/v1/modules/a76/invoices/models.py Normal file
View File

@@ -0,0 +1,259 @@
from enum import Enum
from typing import Optional, List
from sqlalchemy import BigInteger, Boolean, Date, ForeignKey, Integer, Numeric, String, Text, TIMESTAMP
from sqlalchemy.orm import Mapped, mapped_column, relationship
from core.database import Base
from datetime import datetime
from ....common.base_models import TenantScopedMixin, TimestampMixin
class OperationType(str, Enum):
IMP = "imp" # Importación
EXP = "exp" # Exportación
class TransportType(str, Enum):
NONE = "none" # Ninguno
TRANSPORT = "transport" # Transporte
BOX = "box" # Caja
PLATES = "licence plates" # Placas
TRUCK = "truck" # Camión
VESSEL = "vessel" # Buque
BARGE = "rail barge" # Ferrobarcaza
CONTAINER = "container" # Contenedor
AIRPLANE = "airplane" # Avión
GONDOLA = "gondola" # Góndola
FLATBED = "flatbed" # Plataforma
# --- 1. Invoice Header (invoice_header) ---
class InvoiceHeader(Base, TenantScopedMixin, TimestampMixin):
__tablename__ = "invoice_header"
__table_args__ = (
{"schema": "a76"},
)
id: Mapped[int] = mapped_column(
BigInteger, primary_key=True, autoincrement=True)
# Identifiers
operation_type: Mapped[OperationType] = mapped_column(
String(3)) # TIPOMOVIMIENTO / Clasifica imp/exp
invoice_type: Mapped[Optional[str]] = mapped_column(ForeignKey(
"public.invoice_types.key")) # TIPOFACTURA (invoice_types)
invoice_number: Mapped[Optional[str]] = mapped_column(
String(20)) # FACTURAIMPO/FACTURAEXPO/FACTURAREMISION
project_number: Mapped[Optional[str]] = mapped_column(
String(14)) # NUMPROYECTO
purchase_order: Mapped[Optional[str]] = mapped_column(
String(50)) # ORDENCOMPRA
related_doc_id: Mapped[Optional[int]] = mapped_column(
Integer) # IDRELDOC (Para Rectificaciones)
# Dates
invoice_date: Mapped[Optional[datetime]
] = mapped_column(Date) # FECHAFACTURA
capture_date: Mapped[datetime] = mapped_column(
TIMESTAMP(timezone=False), default=datetime.now) # FECHACAPTURA + HORAACTUAL
# Status & Control
is_updated: Mapped[Optional[str]] = mapped_column(Boolean) # ESTATUS
updated_date: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP(timezone=False)) # FECHAACTUALIZACION
who_updated: Mapped[Optional[str]] = mapped_column(String(20)) # QUIENA ACTUALIZO
traffic_light_status: Mapped[Optional[str]] = mapped_column(String(50)) # SEMAFORO / SEMAFOROEXPO/IMPO
process_log: Mapped[Optional[str]] = mapped_column(String(300)) # COMOFUEPROCESADA
# Comments
observation_es: Mapped[Optional[str]] = mapped_column(Text) #OBSERVACIONE
observation_en: Mapped[Optional[str]] = mapped_column(Text) #OBSERVACIONI
comments_status: Mapped[Optional[str]] = mapped_column(Text) #COMENTARIOSESTATUS
# Digital Archive Links
cfdi_uuid: Mapped[Optional[str]] = mapped_column(String(100)) # CFDIUUID
path_pdf: Mapped[Optional[str]] = mapped_column(String(500)) # CFDIPATHPDF
path_xml: Mapped[Optional[str]] = mapped_column(String(500)) # CFDIPATHXML
# Relationships (Para navegacion ORM)
compliance_mx: Mapped["InvoiceComplianceMx"] = relationship(
back_populates="header", cascade="all, delete-orphan")
financials: Mapped["InvoiceFinancials"] = relationship(
back_populates="header", cascade="all, delete-orphan")
details: Mapped[List["InvoiceSalesDetails"]] = relationship(
back_populates="header", cascade="all, delete-orphan")
collections: Mapped[List["InvoiceCollections"]] = relationship(
back_populates="header", cascade="all, delete-orphan")
logistics: Mapped[List["InvoiceLogistics"]] = relationship(
back_populates="header", cascade="all, delete-orphan")
# --- 2. Compliance MX (invoice_compliance_mx) ---
class InvoiceComplianceMx(Base, TenantScopedMixin, TimestampMixin):
__tablename__ = "invoice_compliance_mx"
__table_args__ = (
{"schema": "a76"},
)
invoice_id: Mapped[int] = mapped_column(
ForeignKey("a76.invoice_header.id"), primary_key=True)
# Core Customs Data
pedimento: Mapped[Optional[str]] = mapped_column(
String(19)) # PEDIMENTO/PEDIMENTOIMPO/EXPO
pedimento_code: Mapped[Optional[str]] = mapped_column(
String(5)) # PEDIMENTOR1, K1
remesa: Mapped[Optional[int]] = mapped_column(Integer)
aduana: Mapped[Optional[str]] = mapped_column(ForeignKey("public.customs_sections.customs_code")) # ADUANA_CRUCE
# Clients & Providers
provider_header: Mapped[Optional[str]] = mapped_column(String(20)) # PROVEEDOREXPORTADOR
provider_id: Mapped[Optional[str]] = mapped_column(ForeignKey("a76.clients_and_providers.id")) # PROVEEDOR
sold_to_header: Mapped[Optional[str]] = mapped_column(String(20)) # VENDIDOCONSIGNADO
sold_to_id: Mapped[Optional[str]] = mapped_column(ForeignKey("a76.clients_and_providers.id")) # VENDIDOA
shipped_to_header: Mapped[Optional[str]] = mapped_column(String(20)) # ENVIADOTRANSFERIDO
shipped_to_id: Mapped[Optional[str]] = mapped_column(ForeignKey("a76.clients_and_providers.id")) # ENVIADOA
shipped_by_header: Mapped[Optional[str]] = mapped_column(String(20)) # ENVIADOPORVENDIDOPOR
shipped_by_id: Mapped[Optional[str]] = mapped_column(ForeignKey("a76.clients_and_providers.id")) # VENDIDOPOR
customs_broker_id: Mapped[Optional[str]] = mapped_column(ForeignKey("a76.customs_brokers.id")) # AADUANAL
# Flags & Specific Regimes
is_mixed: Mapped[Optional[bool]] = mapped_column(Boolean) # ESMIXTO
waste_type: Mapped[Optional[str]] = mapped_column(String(1)) # TIPODESPERDICIO
appendix_17: Mapped[Optional[int]] = mapped_column(Integer) # APENDICE17
# VUCEM / Digital
edocument: Mapped[Optional[str]] = mapped_column(String(50)) # EDOCUMENT
electronic_signature: Mapped[Optional[str]] = mapped_column(String(999)) # FIRMAELECTRONICA
sem_id: Mapped[Optional[int]] = mapped_column(Integer) # SEM (de SFacEntradaSM/SFacSalidaSM)
# Relationship
header: Mapped["InvoiceHeader"] = relationship(
back_populates="compliance_mx")
# --- 3. Financials (invoice_financials) ---
class InvoiceFinancials(Base, TenantScopedMixin, TimestampMixin):
__tablename__ = "invoice_financials"
__table_args__ = (
{"schema": "a76"},
)
id: Mapped[int] = mapped_column(
BigInteger, primary_key=True, autoincrement=True)
invoice_id: Mapped[int] = mapped_column(ForeignKey("a76.invoice_header.id"))
currency: Mapped[Optional[str]] = mapped_column(String(3)) # CLAVEMONEDA
currency_type: Mapped[Optional[str]] = mapped_column(ForeignKey("public.currency_types.code")) # TIPOCLAVEMONEDA
exchange_rate: Mapped[Optional[float]] = mapped_column(Numeric(13, 6)) # TIPOCAMBIO
# Merchandise Values
value_mn: Mapped[Optional[float]] = mapped_column(Numeric(23, 8), default=0) # VALORIMPOMN/EXPOMN
value_me: Mapped[Optional[float]] = mapped_column(Numeric(23, 8), default=0)
customs_value_mn: Mapped[Optional[float]] = mapped_column( Numeric(23, 8), default=0) # VALORADUANASMN
# Costs & Taxes
freight: Mapped[Optional[float]] = mapped_column(Numeric(19, 8), default=0) # FLETE
insurance: Mapped[Optional[float]] = mapped_column(Numeric(19, 8), default=0) # VALSEGUROS/SEGUROS
iva_mn: Mapped[Optional[float]] = mapped_column(Numeric(23, 8), default=0) # IVAEXPOMN / VALORIVAMN
iva_factor: Mapped[Optional[float]] = mapped_column(Numeric(17, 4)) # FACTORIVA
# Weights & Quantities
total_quantity: Mapped[Optional[float]] = mapped_column(Numeric(19, 8)) # CANTEXPO / CANTIMPO
gross_weight: Mapped[Optional[float]] = mapped_column(Numeric(19, 8)) # PESOBRUTO
net_weight: Mapped[Optional[float]] = mapped_column(Numeric(19, 8)) # PESONETO
bundle_count: Mapped[Optional[int]] = mapped_column(Integer) # CANTBULTOS
# Relationship
header: Mapped["InvoiceHeader"] = relationship(back_populates="financials")
# --- 4. Logistics (invoice_logistics) ---
class InvoiceLogistics(Base, TenantScopedMixin, TimestampMixin):
__tablename__ = "invoice_logistics"
__table_args__ = (
{"schema": "a76"},
)
id: Mapped[int] = mapped_column(
BigInteger, primary_key=True, autoincrement=True)
invoice_id: Mapped[int] = mapped_column(ForeignKey("a76.invoice_header.id"))
# Carrier Info
carrier_id: Mapped[Optional[str]] = mapped_column(
String(10)) # TRANSPORTISTA
transport_type: Mapped[TransportType] = mapped_column(
String(15), default="none") # TRANSPORTE
transport_mode: Mapped[Optional[str]] = mapped_column(
String(15)) # MODTRANS
driver_name: Mapped[Optional[str]] = mapped_column(String(80)) # CONDUCTOR
is_rail: Mapped[Optional[str]] = mapped_column(String(2)) # ESFERROCARRIL
rail_id: Mapped[Optional[str]] = mapped_column(
String(31)) # IDFERRORCARRIL
# Vehicle & Tracking
vehicle_num: Mapped[Optional[str]] = mapped_column(
String(20)) # NUMVEHICULO / NUMTRAILER
license_plate: Mapped[Optional[str]] = mapped_column(
String(20)) # NUMTRASPORTE
seal_number: Mapped[Optional[str]] = mapped_column(String(15)) # PRECINTO
guide_number: Mapped[Optional[str]] = mapped_column(
String(20)) # NUMEROGUIA
# Logistics Dates
entry_exit_date: Mapped[Optional[datetime]] = mapped_column(
Date) # FECHAENTRADA / FECHAENVIO
# Relationship
header: Mapped["InvoiceHeader"] = relationship(back_populates="logistics")
# --- 5. Sales Order Details (invoice_sales_details) ---
class InvoiceSalesDetails(Base, TenantScopedMixin, TimestampMixin):
__tablename__ = "invoice_sales_details"
__table_args__ = (
{"schema": "a76"},
)
id: Mapped[int] = mapped_column(
BigInteger, primary_key=True, autoincrement=True)
invoice_id: Mapped[int] = mapped_column(ForeignKey("a76.invoice_header.id"))
line_number: Mapped[int] = mapped_column(Integer) # LINEA
sales_order: Mapped[Optional[str]] = mapped_column(
String(20)) # ORDENVENTA
# Specific Custom Fields
colors_description: Mapped[Optional[str]
] = mapped_column(String(49)) # COLORES
square_color_code: Mapped[Optional[str]] = mapped_column(
String(1)) # COLORCUADRITO
line_bundles: Mapped[Optional[int]] = mapped_column(
Integer) # CANTBULTOS (de la linea)
# Relationship
header: Mapped["InvoiceHeader"] = relationship(back_populates="details")
# --- 6. Collections (invoice_collections) ---
class InvoiceCollections(Base, TenantScopedMixin, TimestampMixin):
__tablename__ = "invoice_collections"
__table_args__ = (
{"schema": "a76"},
)
id: Mapped[int] = mapped_column(
BigInteger, primary_key=True, autoincrement=True)
invoice_id: Mapped[int] = mapped_column(ForeignKey("a76.invoice_header.id"))
concept: Mapped[Optional[str]] = mapped_column(String(100)) # CONCEPTO
is_collected: Mapped[Optional[int]] = mapped_column(Integer) # COBRADO
collection_date: Mapped[Optional[datetime]
] = mapped_column(Date) # FECHACOBRANZA
amount: Mapped[Optional[float]] = mapped_column(Numeric(23, 8)) # VALOR
collector_user: Mapped[Optional[str]] = mapped_column(
String(20)) # FACTCOBRADOR
# Relationship
header: Mapped["InvoiceHeader"] = relationship(
back_populates="collections")

284
backend/api/v1/modules/a76/invoices/routes.py Normal file
View File

@@ -0,0 +1,284 @@
from typing import Dict, Any
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from fastapi import APIRouter, Depends, HTTPException, Query, Path
from sqlalchemy.orm import Session
from . import schemas, services
# Create main router
router = APIRouter()
# Create CRUD routes for Invoice Header using TenantCRUDRoutes
invoice_crud = TenantCRUDRoutes(
service=services.InvoiceService,
create_schema=schemas.InvoiceHeaderCreate,
update_schema=schemas.InvoiceHeaderUpdate,
response_schema=schemas.InvoiceHeaderResponse,
prefix="/invoices",
tags=[],
resource_name="Invoice",
id_name="invoice_id",
id_type=int,
enable_list=True, # Enable list endpoint with pagination
enable_filters=True, # Enable filters for status, operation_type, etc.
default_page_size=50,
max_page_size=200,
)
# Include the main CRUD routes
router.include_router(invoice_crud.router)
# Additional nested routes for child resources
# --- Logistics Routes ---
@router.get(
"/invoices/{invoice_id}/logistics",
response_model=list[schemas.InvoiceLogisticsResponse],
summary="Get all logistics for an invoice",
)
def get_invoice_logistics(
invoice_id: int = Path(..., description="Invoice ID"),
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Get all logistics entries for a specific invoice"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
# Verify the invoice exists and belongs to the tenant/company
invoice = services.InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
if not invoice:
raise HTTPException(status_code=404, detail="Invoice not found")
logistics = services.InvoiceLogisticsService.get_all_by_invoice(
db, invoice_id)
return logistics
@router.post(
"/invoices/{invoice_id}/logistics",
response_model=schemas.InvoiceLogisticsResponse,
status_code=201,
summary="Add logistics to an invoice",
)
def create_invoice_logistics(
invoice_id: int = Path(..., description="Invoice ID"),
logistics_data: schemas.InvoiceLogisticsCreate = ...,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Add a new logistics entry to an invoice"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
# Verify the invoice exists and belongs to the tenant/company
invoice = services.InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
if not invoice:
raise HTTPException(status_code=404, detail="Invoice not found")
logistics = services.InvoiceLogisticsService.create(
db, logistics_data, invoice_id, tenant_id, company_id)
return logistics
@router.delete(
"/invoices/{invoice_id}/logistics/{logistics_id}",
status_code=204,
summary="Delete logistics from an invoice",
)
def delete_invoice_logistics(
invoice_id: int = Path(..., description="Invoice ID"),
logistics_id: int = Path(..., description="Logistics ID"),
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Delete a logistics entry from an invoice"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
# Verify the invoice exists and belongs to the tenant/company
invoice = services.InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
if not invoice:
raise HTTPException(status_code=404, detail="Invoice not found")
success = services.InvoiceLogisticsService.delete(
db, logistics_id, invoice_id)
if not success:
raise HTTPException(
status_code=404, detail="Logistics entry not found")
return None
# --- Sales Details Routes ---
@router.get(
"/invoices/{invoice_id}/details",
response_model=list[schemas.InvoiceSalesDetailsResponse],
summary="Get all sales details for an invoice",
)
def get_invoice_details(
invoice_id: int = Path(..., description="Invoice ID"),
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Get all sales details for a specific invoice"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
# Verify the invoice exists and belongs to the tenant/company
invoice = services.InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
if not invoice:
raise HTTPException(status_code=404, detail="Invoice not found")
details = services.InvoiceSalesDetailsService.get_all_by_invoice(
db, invoice_id)
return details
@router.post(
"/invoices/{invoice_id}/details",
response_model=schemas.InvoiceSalesDetailsResponse,
status_code=201,
summary="Add sales detail to an invoice",
)
def create_invoice_detail(
invoice_id: int = Path(..., description="Invoice ID"),
detail_data: schemas.InvoiceSalesDetailsCreate = ...,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Add a new sales detail to an invoice"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
# Verify the invoice exists and belongs to the tenant/company
invoice = services.InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
if not invoice:
raise HTTPException(status_code=404, detail="Invoice not found")
detail = services.InvoiceSalesDetailsService.create(
db, detail_data, invoice_id, tenant_id, company_id)
return detail
@router.delete(
"/invoices/{invoice_id}/details/{detail_id}",
status_code=204,
summary="Delete sales detail from an invoice",
)
def delete_invoice_detail(
invoice_id: int = Path(..., description="Invoice ID"),
detail_id: int = Path(..., description="Detail ID"),
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Delete a sales detail from an invoice"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
# Verify the invoice exists and belongs to the tenant/company
invoice = services.InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
if not invoice:
raise HTTPException(status_code=404, detail="Invoice not found")
success = services.InvoiceSalesDetailsService.delete(
db, detail_id, invoice_id)
if not success:
raise HTTPException(status_code=404, detail="Sales detail not found")
return None
# --- Collections Routes ---
@router.get(
"/invoices/{invoice_id}/collections",
response_model=list[schemas.InvoiceCollectionsResponse],
summary="Get all collections for an invoice",
)
def get_invoice_collections(
invoice_id: int = Path(..., description="Invoice ID"),
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Get all collections for a specific invoice"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
# Verify the invoice exists and belongs to the tenant/company
invoice = services.InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
if not invoice:
raise HTTPException(status_code=404, detail="Invoice not found")
collections = services.InvoiceCollectionsService.get_all_by_invoice(
db, invoice_id)
return collections
@router.post(
"/invoices/{invoice_id}/collections",
response_model=schemas.InvoiceCollectionsResponse,
status_code=201,
summary="Add collection to an invoice",
)
def create_invoice_collection(
invoice_id: int = Path(..., description="Invoice ID"),
collection_data: schemas.InvoiceCollectionsCreate = ...,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Add a new collection to an invoice"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
# Verify the invoice exists and belongs to the tenant/company
invoice = services.InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
if not invoice:
raise HTTPException(status_code=404, detail="Invoice not found")
collection = services.InvoiceCollectionsService.create(
db, collection_data, invoice_id, tenant_id, company_id)
return collection
@router.delete(
"/invoices/{invoice_id}/collections/{collection_id}",
status_code=204,
summary="Delete collection from an invoice",
)
def delete_invoice_collection(
invoice_id: int = Path(..., description="Invoice ID"),
collection_id: int = Path(..., description="Collection ID"),
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Delete a collection from an invoice"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
# Verify the invoice exists and belongs to the tenant/company
invoice = services.InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
if not invoice:
raise HTTPException(status_code=404, detail="Invoice not found")
success = services.InvoiceCollectionsService.delete(
db, collection_id, invoice_id)
if not success:
raise HTTPException(status_code=404, detail="Collection not found")
return None

257
backend/api/v1/modules/a76/invoices/schemas.py Normal file
View File

@@ -0,0 +1,257 @@
from typing import Optional, List
from datetime import datetime, date
from decimal import Decimal
from pydantic import BaseModel, Field
from .models import OperationType
# --- Base Schemas ---
class InvoiceHeaderBase(BaseModel):
"""Base fields for Invoice Header"""
operation_type: Optional[OperationType] = Field(
None, max_length=20, description="Operation type: imp/exp")
invoice_type: Optional[str] = Field(
None, max_length=5, description="Invoice type key")
invoice_number: Optional[str] = Field(
None, max_length=20, description="Invoice number")
project_number: Optional[str] = Field(
None, max_length=14, description="Project number")
purchase_order: Optional[str] = Field(
None, max_length=50, description="Purchase order")
related_doc_id: Optional[int] = Field(
None, description="Related document ID for rectifications")
invoice_date: Optional[date] = Field(None, description="Invoice date")
is_updated: Optional[bool] = Field(None, description="Status")
traffic_light_status: Optional[str] = Field(
None, max_length=50, description="Traffic light status (SEMAFORO)")
process_log: Optional[str] = Field(
None, max_length=300, description="Processing log")
comments_status: Optional[str] = Field(
None, description="Comments and observations")
cfdi_uuid: Optional[str] = Field(
None, max_length=100, description="CFDI UUID")
path_pdf: Optional[str] = Field(
None, max_length=500, description="Path to PDF file")
path_xml: Optional[str] = Field(
None, max_length=500, description="Path to XML file")
class InvoiceComplianceMxBase(BaseModel):
"""Base fields for Compliance MX"""
pedimento: Optional[str] = Field(
None, max_length=19, description="Pedimento number")
pedimento_code: Optional[str] = Field(
None, max_length=5, description="Pedimento code (R1/K1)")
remesa: Optional[int] = Field(None, description="Remesa")
aduana: Optional[str] = Field(
None, max_length=5, description="Customs office")
customs_agent: Optional[str] = Field(
None, max_length=10, description="Customs agent")
is_mixed: Optional[bool] = Field(
None, description="Is mixed operation")
waste_type: Optional[str] = Field(
None, max_length=1, description="Waste type")
appendix_17: Optional[int] = Field(None, description="Appendix 17")
edocument: Optional[str] = Field(
None, max_length=50, description="E-document")
electronic_signature: Optional[str] = Field(
None, max_length=999, description="Electronic signature")
sem_id: Optional[int] = Field(None, description="SEM ID")
class InvoiceFinancialsBase(BaseModel):
"""Base fields for Financials"""
currency: Optional[str] = Field(
None, max_length=3, description="Currency code")
exchange_rate: Optional[Decimal] = Field(None, description="Exchange rate")
value_mn: Optional[Decimal] = Field(None, description="Value in MXN")
value_me: Optional[Decimal] = Field(
None, description="Value in foreign currency")
customs_value_mn: Optional[Decimal] = Field(
None, description="Customs value in MXN")
freight: Optional[Decimal] = Field(None, description="Freight cost")
insurance: Optional[Decimal] = Field(None, description="Insurance cost")
iva_mn: Optional[Decimal] = Field(None, description="IVA in MXN")
iva_factor: Optional[Decimal] = Field(None, description="IVA factor")
total_quantity: Optional[Decimal] = Field(
None, description="Total quantity")
gross_weight: Optional[Decimal] = Field(None, description="Gross weight")
net_weight: Optional[Decimal] = Field(None, description="Net weight")
bundle_count: Optional[int] = Field(None, description="Bundle count")
class InvoiceLogisticsBase(BaseModel):
"""Base fields for Logistics"""
carrier_id: Optional[str] = Field(
None, max_length=10, description="Carrier ID")
transport_mode: Optional[str] = Field(
None, max_length=15, description="Transport mode")
driver_name: Optional[str] = Field(
None, max_length=80, description="Driver name")
is_rail: Optional[str] = Field(
None, max_length=2, description="Is rail transport")
rail_id: Optional[str] = Field(None, max_length=31, description="Rail ID")
vehicle_num: Optional[str] = Field(
None, max_length=20, description="Vehicle number")
license_plate: Optional[str] = Field(
None, max_length=20, description="License plate")
seal_number: Optional[str] = Field(
None, max_length=15, description="Seal number")
guide_number: Optional[str] = Field(
None, max_length=20, description="Guide number")
entry_exit_date: Optional[date] = Field(
None, description="Entry/Exit date")
class InvoiceSalesDetailsBase(BaseModel):
"""Base fields for Sales Details"""
line_number: int = Field(..., description="Line number")
sales_order: Optional[str] = Field(
None, max_length=20, description="Sales order")
colors_description: Optional[str] = Field(
None, max_length=49, description="Colors description")
square_color_code: Optional[str] = Field(
None, max_length=1, description="Square color code")
line_bundles: Optional[int] = Field(None, description="Line bundles count")
class InvoiceCollectionsBase(BaseModel):
"""Base fields for Collections"""
concept: Optional[str] = Field(None, max_length=100, description="Concept")
is_collected: Optional[int] = Field(None, description="Is collected flag")
collection_date: Optional[date] = Field(
None, description="Collection date")
amount: Optional[Decimal] = Field(None, description="Amount")
collector_user: Optional[str] = Field(
None, max_length=20, description="Collector user")
# --- Create Schemas ---
class InvoiceComplianceMxCreate(InvoiceComplianceMxBase):
"""Schema for creating Compliance MX"""
pass
class InvoiceFinancialsCreate(InvoiceFinancialsBase):
"""Schema for creating Financials"""
pass
class InvoiceLogisticsCreate(InvoiceLogisticsBase):
"""Schema for creating Logistics"""
pass
class InvoiceSalesDetailsCreate(InvoiceSalesDetailsBase):
"""Schema for creating Sales Details"""
pass
class InvoiceCollectionsCreate(InvoiceCollectionsBase):
"""Schema for creating Collections"""
pass
class InvoiceHeaderCreate(InvoiceHeaderBase):
"""Schema for creating Invoice Header with nested relations"""
compliance_mx: Optional[InvoiceComplianceMxCreate] = None
financials: Optional[InvoiceFinancialsCreate] = None
logistics: Optional[List[InvoiceLogisticsCreate]] = None
details: Optional[List[InvoiceSalesDetailsCreate]] = None
collections: Optional[List[InvoiceCollectionsCreate]] = None
# --- Update Schemas ---
class InvoiceComplianceMxUpdate(InvoiceComplianceMxBase):
"""Schema for updating Compliance MX"""
pass
class InvoiceFinancialsUpdate(InvoiceFinancialsBase):
"""Schema for updating Financials"""
pass
class InvoiceLogisticsUpdate(InvoiceLogisticsBase):
"""Schema for updating Logistics"""
pass
class InvoiceSalesDetailsUpdate(InvoiceSalesDetailsBase):
"""Schema for updating Sales Details"""
line_number: Optional[int] = None
class InvoiceCollectionsUpdate(InvoiceCollectionsBase):
"""Schema for updating Collections"""
pass
class InvoiceHeaderUpdate(InvoiceHeaderBase):
"""Schema for updating Invoice Header with nested relations"""
compliance_mx: Optional[InvoiceComplianceMxUpdate] = None
financials: Optional[InvoiceFinancialsUpdate] = None
logistics: Optional[List[InvoiceLogisticsUpdate]] = None
details: Optional[List[InvoiceSalesDetailsUpdate]] = None
collections: Optional[List[InvoiceCollectionsUpdate]] = None
# --- Response Schemas ---
class InvoiceComplianceMxResponse(InvoiceComplianceMxBase):
"""Schema for Compliance MX response"""
invoice_id: int
class Config:
from_attributes = True
class InvoiceFinancialsResponse(InvoiceFinancialsBase):
"""Schema for Financials response"""
invoice_id: int
class Config:
from_attributes = True
class InvoiceLogisticsResponse(InvoiceLogisticsBase):
"""Schema for Logistics response"""
logistics_id: int
invoice_id: int
class Config:
from_attributes = True
class InvoiceSalesDetailsResponse(InvoiceSalesDetailsBase):
"""Schema for Sales Details response"""
detail_id: int
invoice_id: int
class Config:
from_attributes = True
class InvoiceCollectionsResponse(InvoiceCollectionsBase):
"""Schema for Collections response"""
collection_id: int
invoice_id: int
class Config:
from_attributes = True
class InvoiceHeaderResponse(InvoiceHeaderBase):
"""Schema for Invoice Header response with nested relations"""
id: int
capture_date: datetime
compliance_mx: Optional[InvoiceComplianceMxResponse] = None
financials: Optional[InvoiceFinancialsResponse] = None
logistics: List[InvoiceLogisticsResponse] = []
details: List[InvoiceSalesDetailsResponse] = []
collections: List[InvoiceCollectionsResponse] = []
class Config:
from_attributes = True

256
backend/api/v1/modules/a76/invoices/services.py Normal file
View File

@@ -0,0 +1,256 @@
import traceback
from typing import Optional, List, Tuple
from sqlalchemy.orm import Session
from sqlalchemy import and_
from . import models, schemas
class InvoiceService:
"""Service for Invoice Header operations"""
@staticmethod
def get_by_id(db: Session, invoice_id: int, tenant_id: int, company_id: int) -> Optional[models.InvoiceHeader]:
"""Get an invoice by ID with tenant/company validation"""
return (
db.query(models.InvoiceHeader)
.filter(
models.InvoiceHeader.id == invoice_id,
models.InvoiceHeader.tenant_id == tenant_id,
models.InvoiceHeader.company_id == company_id,
)
.first()
)
@staticmethod
def get_all(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 100,
filters: Optional[dict] = None,
) -> Tuple[List[models.InvoiceHeader], int]:
"""Get all invoices for a tenant/company with pagination and optional filters"""
query = db.query(models.InvoiceHeader).filter(
models.InvoiceHeader.tenant_id == tenant_id,
models.InvoiceHeader.company_id == company_id,
)
# Apply filters if provided
if filters:
if filters.get("status"):
query = query.filter(
models.InvoiceHeader.status == filters["status"])
if filters.get("operation_type"):
query = query.filter(
models.InvoiceHeader.operation_type == filters["operation_type"])
if filters.get("invoice_type"):
query = query.filter(
models.InvoiceHeader.invoice_type == filters["invoice_type"])
if filters.get("invoice_number"):
query = query.filter(models.InvoiceHeader.invoice_number.ilike(
f"%{filters['invoice_number']}%"))
if filters.get("pedimento"):
query = query.join(models.InvoiceComplianceMx).filter(
models.InvoiceComplianceMx.pedimento.ilike(
f"%{filters['pedimento']}%")
)
if not filters.get("invoice_type") and filters.get("operation_type") == "exp":
query = query.filter(
models.InvoiceHeader.operation_type != "REPAR")
total = query.count()
items = query.offset(skip).limit(limit).all()
return items, total
@staticmethod
def create(
db: Session,
invoice_data: schemas.InvoiceHeaderCreate,
tenant_id: int,
company_id: int
) -> models.InvoiceHeader:
"""Create a new invoice with all related data"""
def clean_dict(data_dict: dict) -> dict:
cleaned = {}
for key, value in data_dict.items():
if key == 'customs_agent':
key = 'customs_broker_id'
elif key == 'provider':
key = 'provider_id'
if isinstance(value, str) and not value.strip():
cleaned[key] = None
elif value == 0 and (key.endswith('_id') or key == 'remesa'):
cleaned[key] = None
else:
cleaned[key] = value
return cleaned
try:
# Extract nested data
compliance_data = invoice_data.compliance_mx
financials_data = invoice_data.financials
logistics_data = invoice_data.logistics or []
details_data = invoice_data.details or []
collections_data = invoice_data.collections or []
# Create main invoice header
raw_invoice_dict = invoice_data.model_dump(
exclude={"compliance_mx", "financials",
"logistics", "details", "collections"}
)
invoice_dict = clean_dict(raw_invoice_dict)
invoice_dict["tenant_id"] = tenant_id
invoice_dict["company_id"] = company_id
new_invoice = models.InvoiceHeader(**invoice_dict)
db.add(new_invoice)
db.flush() # Flush to get the invoice ID
# Create compliance_mx if provided
if compliance_data:
raw_comp_dict = compliance_data.model_dump()
# Pasamos los datos por la lavadora para arreglar pedimento, aduana, etc.
compliance_dict = clean_dict(raw_comp_dict)
compliance_dict["invoice_id"] = new_invoice.id
compliance_dict["tenant_id"] = tenant_id
compliance_dict["company_id"] = company_id
new_compliance = models.InvoiceComplianceMx(**compliance_dict)
db.add(new_compliance)
# Create financials if provided
if financials_data:
raw_fin_dict = financials_data.model_dump()
financials_dict = clean_dict(raw_fin_dict)
financials_dict["invoice_id"] = new_invoice.id
financials_dict["tenant_id"] = tenant_id
financials_dict["company_id"] = company_id
new_financials = models.InvoiceFinancials(**financials_dict)
db.add(new_financials)
# Create logistics entries
for logistics_item in logistics_data:
raw_log_dict = logistics_item.model_dump()
logistics_dict = clean_dict(raw_log_dict)
logistics_dict["invoice_id"] = new_invoice.id
logistics_dict["tenant_id"] = tenant_id
logistics_dict["company_id"] = company_id
new_logistics = models.InvoiceLogistics(**logistics_dict)
db.add(new_logistics)
# Create sales details
for detail_item in details_data:
raw_det_dict = detail_item.model_dump()
detail_dict = clean_dict(raw_det_dict)
detail_dict["invoice_id"] = new_invoice.id
detail_dict["tenant_id"] = tenant_id
detail_dict["company_id"] = company_id
new_detail = models.InvoiceSalesDetails(**detail_dict)
db.add(new_detail)
# Create collections
for collection_item in collections_data:
raw_col_dict = collection_item.model_dump()
collection_dict = clean_dict(raw_col_dict)
collection_dict["invoice_id"] = new_invoice.id
collection_dict["tenant_id"] = tenant_id
collection_dict["company_id"] = company_id
new_collection = models.InvoiceCollections(**collection_dict)
db.add(new_collection)
db.commit()
db.refresh(new_invoice)
return new_invoice
except Exception as e:
db.rollback()
print("\n\n🔥 ERROR AL GUARDAR FACTURA 🔥")
print(f"Error: {str(e)}")
traceback.print_exc() # Esto imprime el error real en la consola
print("--------------------------------\n")
raise e
@staticmethod
def update(
db: Session,
invoice_id: int,
tenant_id: int,
invoice_data: schemas.InvoiceHeaderUpdate,
company_id: int
) -> Optional[models.InvoiceHeader]:
# ... (El resto de tu código update se queda igual) ...
# (Te recomiendo implementar clean_dict aquí también si tienes problemas al editar)
invoice = InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
if not invoice:
return None
# Update main invoice header fields
update_dict = invoice_data.model_dump(
exclude={"compliance_mx", "financials",
"logistics", "details", "collections"},
exclude_unset=True
)
for key, value in update_dict.items():
setattr(invoice, key, value)
# Update compliance_mx if provided
if invoice_data.compliance_mx is not None:
if invoice.compliance_mx:
for key, value in invoice_data.compliance_mx.model_dump(exclude_unset=True).items():
# Parche rápido para update
if value == "": value = None
setattr(invoice.compliance_mx, key, value)
else:
compliance_dict = invoice_data.compliance_mx.model_dump()
# Aplicar limpieza manual si es necesario
if 'customs_agent' in compliance_dict: compliance_dict['customs_broker_id'] = compliance_dict.pop('customs_agent')
compliance_dict["invoice_id"] = invoice.id
compliance_dict["tenant_id"] = tenant_id
compliance_dict["company_id"] = company_id
new_compliance = models.InvoiceComplianceMx(**compliance_dict)
db.add(new_compliance)
# Update financials if provided
if invoice_data.financials is not None:
if invoice.financials:
for key, value in invoice_data.financials.model_dump(exclude_unset=True).items():
if value == "": value = None
setattr(invoice.financials, key, value)
else:
financials_dict = invoice_data.financials.model_dump()
financials_dict["invoice_id"] = invoice.id
financials_dict["tenant_id"] = tenant_id
financials_dict["company_id"] = company_id
new_financials = models.InvoiceFinancials(**financials_dict)
db.add(new_financials)
db.commit()
db.refresh(invoice)
return invoice
@staticmethod
def delete(db: Session, invoice_id: int, tenant_id: int, company_id: int) -> bool:
"""Delete an invoice and all related data (cascade delete)"""
invoice = InvoiceService.get_by_id(
db, invoice_id, tenant_id, company_id)
if invoice:
db.delete(invoice)
db.commit()
return True
return False

11
backend/api/v1/modules/a76/router.py
View File

@@ -8,7 +8,7 @@ from fastapi import APIRouter
from .customs_brokers.routes import router as customs_broker_router
# Importar routers de módulos
from ..core.auth import router as auth_router
from .invoices.routes import router as invoices_router
from .classes import router as classes_router
from .clients_and_providers import router as client_and_provider_router
from .general_catalogs.company import router as company_router
@@ -17,7 +17,6 @@ from .transportation.drivers.routes import router as drivers_router
from .general_catalogs.exchange_rate.routes import router as exchange_rate_router
from .general_catalogs.identifiers.routes import router as identifiers_router
from .fraction_rule_octave.routes import router as fraction_rule_octave_router
from ..core.licenses import router as licenses_router
from .general_catalogs.packages.routes import router as package_router
from .general_catalogs.ports.routes import router as ports_router
from .parts import router as parts_router
@@ -38,21 +37,15 @@ from .general_catalogs.error_catalogs.routes import router as error_catalogs_rou
from .general_catalogs.doda.routes import router as doda_router
from .general_catalogs.prevalidators.routes import router as prevalidators_router
from .general_catalogs.electronic_notices.routes import router as electronic_notices_router
from ..core.tenants import router as tenants_router
from .transportation.trailers.routes import router as trailers_router
from .transportation.transporters.routes import router as transporters_router
from ..core.user_tenant.routes import router as user_tenant_router
from .transportation.vehicles.routes import router as vehicles_router
# Router principal
router = APIRouter()
# Registrar módulos
router.include_router(auth_router)
router.include_router(tenants_router, prefix="/a76", tags=["a76 / tenants"])
router.include_router(user_tenant_router, prefix="/a76",
tags=["a76 / user-tenants"])
router.include_router(licenses_router, prefix="/a76", tags=["a76 / licenses"])
router.include_router(invoices_router, prefix="/a76", tags=["a76 / invoices"])
router.include_router(pedimentos_router, prefix="/a76")
router.include_router(
client_and_provider_router, prefix="/a76", tags=["a76 / clients_and_providers"]