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

Desarrollo de rutina de CRUD para los catalogos generales

Browse Source
This commit is contained in:
Kevin_Ramirez
2025-12-22 17:43:25 -06:00
parent 6a9f69e39e
commit f915bccd5f
95 changed files with 4414 additions and 1137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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."
)

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

@@ -40,6 +40,55 @@ async def create_company(
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],
@@ -189,16 +238,89 @@ async def get_program_info(
"prosec_authorization": company.prosec_authorization,
}
base_router = TenantCRUDRoutes(
service=CompanyService,
create_schema=None,
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

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

@@ -137,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]:

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")

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")

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)}")

10
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="/legends",
tags=["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."
)

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")

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):