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

feat: Enhance Pedimentos creation and validation logic in frontend and backend

Browse Source
This commit is contained in:
acazares
2025-11-12 10:42:15 -06:00
parent a5c1bab201
commit 67a309912c
6 changed files with 223 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
backend/api/v1/common/tenant_crud_routes.py
View File

@@ -2,7 +2,7 @@ from typing import Any, Callable, Dict, Generic, Optional, Type, TypeVar, Union
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from fastapi import APIRouter, Depends, HTTPException, Path, Query
from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query
from pydantic import BaseModel
from sqlalchemy.orm import Session
@@ -113,7 +113,12 @@ class TenantCRUDRoutes(
if self.enable_list:
if self.enable_filters:
@self.router.get("/", response_model=Dict[str, Any])
@self.router.get(
"/",
response_model=Dict[str, Any],
summary=f"List {self.resource_name}s",
description=f"Get paginated list of {self.resource_name}s with optional filters",
)
async def list_resources(
company_id: int = Query(..., description="Company ID"),
page: int = Query(1, ge=1, description="Page number"),
@@ -127,7 +132,6 @@ class TenantCRUDRoutes(
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
):
f"""List all {self.resource_name}s with pagination"""
tenant_id = validate_access_to_resource(
db, company_id, current_user
)
@@ -152,7 +156,12 @@ class TenantCRUDRoutes(
else:
@self.router.get("/", response_model=Dict[str, Any])
@self.router.get(
"/",
response_model=Dict[str, Any],
summary=f"List {self.resource_name}s",
description=f"Get paginated list of {self.resource_name}s",
)
async def list_resources(
company_id: int = Query(..., description="Company ID"),
page: int = Query(1, ge=1, description="Page number"),
@@ -165,7 +174,6 @@ class TenantCRUDRoutes(
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
):
f"""List all {self.resource_name}s with pagination"""
tenant_id = validate_access_to_resource(
db, company_id, current_user
)
@@ -190,14 +198,19 @@ class TenantCRUDRoutes(
# For child resources: GET / (parent_id comes from path)
if self.parent_id_name:
# Child resource - single GET without ID in path
@self.router.get("/", response_model=self.response_schema)
@self.router.get(
"/",
response_model=self.response_schema,
summary=f"Get {self.resource_name}",
description=f"Get {self.resource_name} by {self.parent_id_name}",
)
async def get_resource(
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
**path_params,
):
f"""Get {self.resource_name} by {self.parent_id_name}"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
parent_id = path_params.get(self.parent_id_name)
@@ -223,15 +236,19 @@ class TenantCRUDRoutes(
else:
# Parent resource - GET by ID in path
@self.router.get(
f"/{{{self.id_name}}}", response_model=self.response_schema
f"/{{{self.id_name}}}",
response_model=self.response_schema,
summary=f"Get {self.resource_name} by ID",
description=f"Get a specific {self.resource_name} by {self.id_name}",
)
async def get_resource_by_id(
resource_id: Union[int, str] = Path(..., alias=self.id_name),
resource_id: Union[int, str] = Path(
..., alias=self.id_name, description=f"{self.resource_name} ID"
),
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
):
f"""Get {self.resource_name} by ID"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
resource = self.service.get_by_id(
@@ -245,49 +262,80 @@ class TenantCRUDRoutes(
return resource
# POST route
@self.router.post("/", response_model=self.response_schema, status_code=201)
async def create_resource(
data: CreateSchemaType,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
**path_params,
):
f"""Create {self.resource_name}"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
# Validate parent ID match if enabled and parent_id_name exists
if self.validate_parent_match and self.parent_id_name:
parent_id = path_params.get(self.parent_id_name)
data_parent_id = getattr(data, self.parent_id_name, None)
if data_parent_id is not None and data_parent_id != parent_id:
raise HTTPException(
status_code=400,
detail=f"{self.parent_id_name.replace('_', ' ').title()} mismatch",
)
resource = self.service.create(db, data, tenant_id, company_id)
return resource
if self.parent_id_name:
# Child resource - needs parent_id from path
# Create a closure to capture the schema type
create_schema = self.create_schema
@self.router.post(
"/",
response_model=self.response_schema,
status_code=201,
summary=f"Create {self.resource_name}",
description=f"Create a new {self.resource_name}",
)
async def create_child_resource(
company_id: int = Query(..., description="Company ID"),
data: create_schema = Body(...), # type: ignore
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
):
tenant_id = validate_access_to_resource(db, company_id, current_user)
# For child resources, parent_id validation would go here
resource = self.service.create(db, data, tenant_id, company_id)
return resource
else:
# Parent resource - no parent_id needed
# Create a closure to capture the schema type
create_schema = self.create_schema
@self.router.post(
"/",
response_model=self.response_schema,
status_code=201,
summary=f"Create {self.resource_name}",
description=f"Create a new {self.resource_name}",
)
async def create_parent_resource(
company_id: int = Query(..., description="Company ID"),
data: create_schema = Body(...), # type: ignore
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
):
tenant_id = validate_access_to_resource(db, company_id, current_user)
resource = self.service.create(db, data, tenant_id, company_id)
return resource
# PUT route
# For parent resources: PUT /{id}
# For child resources: PUT / (parent_id comes from path)
if self.parent_id_name:
# Child resource
@self.router.put("/", response_model=self.response_schema)
# Create a closure to capture the schema type
update_schema = self.update_schema
@self.router.put(
"/",
response_model=self.response_schema,
summary=f"Update {self.resource_name}",
description=f"Update an existing {self.resource_name}",
)
async def update_resource(
data: UpdateSchemaType,
company_id: int = Query(..., description="Company ID"),
data: update_schema = Body(...), # type: ignore
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
**path_params,
):
f"""Update {self.resource_name}"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
parent_id = path_params.get(self.parent_id_name)
resource = self.service.update(
db, parent_id, tenant_id, company_id, data
db, parent_id, tenant_id, data, company_id
)
if not resource:
@@ -298,13 +346,22 @@ class TenantCRUDRoutes(
else:
# Parent resource
# Create a closure to capture the schema type
update_schema = self.update_schema
@self.router.put(
f"/{{{self.id_name}}}", response_model=self.response_schema
f"/{{{self.id_name}}}",
response_model=self.response_schema,
summary=f"Update {self.resource_name}",
description=f"Update an existing {self.resource_name} by {self.id_name}",
)
async def update_resource_by_id(
data: UpdateSchemaType,
resource_id: Union[int, str] = Path(..., alias=self.id_name),
resource_id: Union[int, str] = Path(
..., alias=self.id_name, description=f"{self.resource_name} ID"
),
company_id: int = Query(..., description="Company ID"),
data: update_schema = Body(...), # type: ignore
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
):
@@ -312,7 +369,7 @@ class TenantCRUDRoutes(
tenant_id = validate_access_to_resource(db, company_id, current_user)
resource = self.service.update(
db, resource_id, tenant_id, company_id, data
db, resource_id, tenant_id, data, company_id
)
if not resource:
@@ -326,14 +383,18 @@ class TenantCRUDRoutes(
# For child resources: DELETE / (parent_id comes from path)
if self.parent_id_name:
# Child resource
@self.router.delete("/", status_code=204)
@self.router.delete(
"/",
status_code=204,
summary=f"Delete {self.resource_name}",
description=f"Delete an existing {self.resource_name}",
)
async def delete_resource(
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
**path_params,
):
f"""Delete {self.resource_name}"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
parent_id = path_params.get(self.parent_id_name)
@@ -347,14 +408,20 @@ class TenantCRUDRoutes(
else:
# Parent resource
@self.router.delete(f"/{{{self.id_name}}}", status_code=204)
@self.router.delete(
f"/{{{self.id_name}}}",
status_code=204,
summary=f"Delete {self.resource_name}",
description=f"Delete an existing {self.resource_name} by {self.id_name}",
)
async def delete_resource_by_id(
resource_id: Union[int, str] = Path(..., alias=self.id_name),
resource_id: Union[int, str] = Path(
..., alias=self.id_name, description=f"{self.resource_name} ID"
),
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(self.db_dependency),
current_user: Dict[str, Any] = Depends(self.auth_dependency),
):
f"""Delete {self.resource_name}"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
success = self.service.delete(db, resource_id, tenant_id, company_id)

11
backend/api/v1/modules/a76/pedmientos/dtos/pedimentos.py
View File

@@ -39,7 +39,16 @@ class PedimentosBase(BaseModel):
class PedimentosCreate(PedimentosBase):
"""Schema for creating a new Pedimento"""
pass
# Override to make required fields non-optional
year: str = Field(..., max_length=2, description="Year")
customs_office: str = Field(..., max_length=2, description="Customs office")
license: str = Field(..., max_length=4, description="License")
pedimento_number: str = Field(..., max_length=7, description="Pedimento number")
client_id: int = Field(..., description="Client ID")
operation_type: int = Field(..., description="Operation type")
pedimento_type: int = Field(..., description="Pedimento type")
regime: str = Field(..., max_length=3, description="Regime")
status: str = Field(..., max_length=30, description="Status")
class PedimentosUpdate(BaseModel):

2
backend/api/v1/modules/a76/pedmientos/routes/pedimentos.py
View File

@@ -14,7 +14,7 @@ router = TenantCRUDRoutes(
update_schema=PedimentosUpdate,
response_schema=PedimentosResponse,
prefix="", # No prefix here, will be added in main router
tags=[],
tags=["a76 / pedimentos"], # Tag for Swagger documentation
resource_name="Pedimento",
id_name="pedimento_id",
enable_list=True, # Enable GET / with pagination

33
backend/api/v1/modules/a76/pedmientos/services/pedimentos.py
View File

@@ -55,7 +55,7 @@ class PedimentosService:
@staticmethod
def get_by_id(
db: Session, pedimento_id: int, tenant_id: int
db: Session, pedimento_id: int, tenant_id: int, company_id: int = None
) -> Optional[Pedimentos]:
"""
Get a pedimento by ID
@@ -64,19 +64,23 @@ class PedimentosService:
db: Database session
pedimento_id: Pedimento ID
tenant_id: Tenant ID
company_id: Company ID (optional for backwards compatibility)
Returns:
Pedimento or None if not found
"""
return (
db.query(Pedimentos)
.filter(Pedimentos.id == pedimento_id, Pedimentos.tenant_id == tenant_id)
.first()
query = db.query(Pedimentos).filter(
Pedimentos.id == pedimento_id, Pedimentos.tenant_id == tenant_id
)
if company_id is not None:
query = query.filter(Pedimentos.company_id == company_id)
return query.first()
@staticmethod
def create(
db: Session, pedimento_data: PedimentosCreate, tenant_id: int
db: Session, pedimento_data: PedimentosCreate, tenant_id: int, company_id: int
) -> Pedimentos:
"""
Create a new pedimento
@@ -84,12 +88,15 @@ class PedimentosService:
Args:
db: Database session
pedimento_data: Pedimento creation data
tenant_id: Tenant ID
company_id: Company ID
Returns:
Created pedimento
"""
pedimento = Pedimentos(**pedimento_data.model_dump())
pedimento.tenant_id = 1
pedimento.tenant_id = tenant_id
pedimento.company_id = company_id
db.add(pedimento)
db.commit()
@@ -98,7 +105,7 @@ class PedimentosService:
@staticmethod
def update(
db: Session, pedimento_id: int, tenant_id: int, pedimento_data: PedimentosUpdate
db: Session, pedimento_id: int, tenant_id: int, pedimento_data: PedimentosUpdate, company_id: int = None
) -> Optional[Pedimentos]:
"""
Update a pedimento
@@ -108,11 +115,12 @@ class PedimentosService:
pedimento_id: Pedimento ID
tenant_id: Tenant ID
pedimento_data: Updated data
company_id: Company ID (optional for backwards compatibility)
Returns:
Updated pedimento or None if not found
"""
pedimento = PedimentosService.get_by_id(db, pedimento_id, tenant_id)
pedimento = PedimentosService.get_by_id(db, pedimento_id, tenant_id, company_id)
if not pedimento:
return None
@@ -125,7 +133,7 @@ class PedimentosService:
return pedimento
@staticmethod
def delete(db: Session, pedimento_id: int, tenant_id: int) -> bool:
def delete(db: Session, pedimento_id: int, tenant_id: int, company_id: int = None) -> bool:
"""
Delete a pedimento
@@ -133,14 +141,17 @@ class PedimentosService:
db: Database session
pedimento_id: Pedimento ID
tenant_id: Tenant ID
company_id: Company ID (optional for backwards compatibility)
Returns:
True if deleted, False if not found
"""
pedimento = PedimentosService.get_by_id(db, pedimento_id, tenant_id)
pedimento = PedimentosService.get_by_id(db, pedimento_id, tenant_id, company_id)
if not pedimento:
return False
db.delete(pedimento)
db.commit()
return True
db.commit()
return True