feat: Implement consolidated catalog services and endpoints for pedimento and invoice creation/edition.

This commit is contained in:
Galindo97
2026-02-12 13:35:48 -06:00
parent 9d27066da1
commit ab5181f803
16 changed files with 1323 additions and 879 deletions

View File

@@ -196,4 +196,33 @@ class ClassSearchDTO(BaseModel):
)
class Config:
from_attributes = True
from_attributes = True
class ClassWithFADataResponse(BaseModel):
"""DTO para respuesta de clase con datos FA embebidos (para fixed-asset-classes)"""
# Base class fields
id: int
tenant_id: int
company_id: int
class_code: str
description_es: Optional[str] = None
description_en: Optional[str] = None
material_key: Optional[str] = None
unit_of_measure: Optional[str] = None
fraction: Optional[str] = None
us_fraction: Optional[str] = None
sub_key: Optional[str] = None
physical_review: Optional[int] = None
iva_exempt_fraction: Optional[str] = None
created_at: datetime
updated_at: datetime
# FA-specific fields (embedded from a24.fa_classes)
fa_class_id: Optional[int] = None
depreciation_rate: Optional[Decimal] = None
fda_code: Optional[str] = None
class_enabled: Optional[bool] = None
model_config = ConfigDict(from_attributes=True)

View File

@@ -2,34 +2,53 @@
Endpoints API para gestión de clases SCAII y SCAF
"""
from typing import Dict, Any
from fastapi import Depends, Query
from typing import Dict, Any, List
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from core.database import get_core_db
from core.security import get_current_user
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes, validate_access_to_resource
from .dto import ClassCreateDTO, ClassCreateDTOFA, ClassResponseDTO, ClassResponseDTOFA, ClassUpdateDTO
from .dto import ClassCreateDTO, ClassCreateDTOFA, ClassResponseDTO, ClassResponseDTOFA, ClassUpdateDTO, ClassWithFADataResponse
from .service import ClassService
# Create router with generic CRUD routes
crud_routes = TenantCRUDRoutes(
service=ClassService,
create_schema=ClassCreateDTO,
update_schema=ClassUpdateDTO,
response_schema=ClassResponseDTO,
prefix="/classes",
tags=["a76 / classes"],
resource_name="Class",
id_name="id",
enable_list=True,
enable_filters=True,
default_page_size=50,
max_page_size=1000,
)
# Create a new router for custom endpoints
router = APIRouter()
router = crud_routes.router
# Add consolidated catalog endpoints FIRST (before generic CRUD routes)
# This ensures they have priority over the generic /{id} route
@router.get(
"/with-fa-data",
response_model=List[ClassWithFADataResponse],
summary="Get Classes with FA Data",
description="Get all classes with their FA data in a single query (eliminates N+1 problem)",
tags=["a76 / classes"],
)
async def get_classes_with_fa_data(
company_id: int = Query(..., description="Company ID"),
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(1000, ge=1, le=1000, description="Page size"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""
Get all classes with their FA data using a single LEFT JOIN query.
This endpoint is optimized for the fixed-asset-classes view.
"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
skip = (page - 1) * page_size
classes_with_fa, total = ClassService.get_all_with_fa_data(
db=db,
tenant_id=tenant_id,
company_id=company_id,
skip=skip,
limit=page_size,
)
return classes_with_fa
@router.post(
"/fa",
@@ -37,6 +56,7 @@ router = crud_routes.router
status_code=201,
summary="Create Fixed Asset Class",
description="Create a class with FA extension in a single transaction",
tags=["a76 / classes"],
)
async def create_fa_class(
class_data: ClassCreateDTOFA,
@@ -50,4 +70,24 @@ async def create_fa_class(
result = ClassService.create_fa_class(db, class_data, tenant_id, company_id)
return result
return result
# Now include generic CRUD routes
# These will be registered AFTER the custom endpoints above
crud_router = TenantCRUDRoutes(
service=ClassService,
create_schema=ClassCreateDTO,
update_schema=ClassUpdateDTO,
response_schema=ClassResponseDTO,
prefix="", # No prefix here, will be added in main router
tags=["a76 / classes"],
resource_name="Class",
id_name="id",
enable_list=True,
enable_filters=True,
default_page_size=50,
max_page_size=1000,
).router
# Include the CRUD routes into our main router
router.include_router(crud_router)

View File

@@ -73,6 +73,91 @@ class ClassService:
return items, total
@staticmethod
def get_all_with_fa_data(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 1000,
filters: Optional[Dict[str, Any]] = None,
) -> tuple[List[Dict[str, Any]], int]:
"""
Get all classes with their FA data in a single query using LEFT JOIN.
This eliminates the N+1 query problem.
Returns a list of dicts with combined base class + FA data.
"""
from api.v1.modules.a24.fa.fa_classes.models import QClasses
# Build query with LEFT JOIN
query = (
db.query(Class, QClasses)
.outerjoin(QClasses, and_(
Class.id == QClasses.class_id,
QClasses.tenant_id == tenant_id
))
.filter(Class.tenant_id == tenant_id)
.filter(Class.company_id == company_id)
)
# Apply filters if provided
if filters:
if filters.get("class_code"):
query = query.filter(
Class.class_code.ilike(f"%{filters['class_code']}%")
)
if filters.get("description"):
description_pattern = f"%{filters['description']}%"
query = query.filter(
or_(
Class.description_es.ilike(description_pattern),
Class.description_en.ilike(description_pattern),
)
)
if filters.get("material_key"):
query = query.filter(
Class.material_key.ilike(f"%{filters['material_key']}%")
)
if filters.get("fraction"):
query = query.filter(Class.fraction.ilike(f"%{filters['fraction']}%"))
# Count total before pagination
total = query.count()
# Apply pagination
results = query.offset(skip).limit(limit).all()
# Combine base class + FA data into dicts
combined = []
for base_class, fa_class in results:
class_dict = {
# Base class fields
"id": base_class.id,
"tenant_id": base_class.tenant_id,
"company_id": base_class.company_id,
"class_code": base_class.class_code,
"description_es": base_class.description_es,
"description_en": base_class.description_en,
"material_key": base_class.material_key,
"unit_of_measure": base_class.unit_of_measure,
"fraction": base_class.fraction,
"us_fraction": base_class.us_fraction,
"sub_key": base_class.sub_key,
"physical_review": base_class.physical_review,
"iva_exempt_fraction": base_class.iva_exempt_fraction,
"created_at": base_class.created_at,
"updated_at": base_class.updated_at,
# FA extension fields (None if no FA record exists)
"fa_class_id": fa_class.id if fa_class else None,
"depreciation_rate": fa_class.depreciation_rate if fa_class else None,
"fda_code": fa_class.fda_code if fa_class else None,
"class_enabled": fa_class.class_enabled if fa_class else None,
}
combined.append(class_dict)
return combined, total
@staticmethod
def get_by_id(
db: Session, class_id: int, tenant_id: int, company_id: int