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

Merge pull request 'feat: Convertir módulo de clases para usar TenantCRUDRoutes' (#20) from classes into development

Browse Source 
Reviewed-on: ADUANASOFT/anexo76#20
This commit is contained in:
acazares
2025-11-18 22:36:29 +00:00
parent 243fab8bef 99aff32eea
commit 84006e3410
14 changed files with 1702 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
backend/api/v1/modules/a76/classes/dto.py
View File

@@ -3,9 +3,10 @@ DTOs (Data Transfer Objects) para módulo de clases SCAII y SCAF
Reemplaza schemas.py siguiendo enfoque DDD y estilo NestJS
"""
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field
from pydantic import BaseModel, ConfigDict, Field
class ClassCreateDTO(BaseModel):
@@ -13,10 +14,10 @@ class ClassCreateDTO(BaseModel):
client_id: int = Field(..., description="Client key")
class_code: str = Field(..., max_length=8, description="Class code")
description_spanish: Optional[str] = Field(
description_es: Optional[str] = Field(
None, max_length=500, description="Description in Spanish"
)
description_english: Optional[str] = Field(
description_en: Optional[str] = Field(
None, max_length=500, description="Description in English"
)
material_key: Optional[str] = Field(
@@ -50,10 +51,10 @@ class ClassCreateDTO(BaseModel):
class ClassUpdateDTO(BaseModel):
"""DTO para actualizar una clase"""
description_spanish: Optional[str] = Field(
description_es: Optional[str] = Field(
None, max_length=500, description="Description in Spanish"
)
description_english: Optional[str] = Field(
description_en: Optional[str] = Field(
None, max_length=500, description="Description in English"
)
material_key: Optional[str] = Field(
@@ -87,10 +88,13 @@ class ClassUpdateDTO(BaseModel):
class ClassResponseDTO(BaseModel):
"""DTO para respuesta de clase"""
id: int
tenant_id: int
company_id: int
client_id: int
class_code: str
description_spanish: Optional[str] = None
description_english: Optional[str] = None
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
@@ -98,9 +102,10 @@ class ClassResponseDTO(BaseModel):
sub_key: Optional[str] = None
physical_review: Optional[int] = None
iva_exempt_fraction: Optional[str] = None
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
model_config = ConfigDict(from_attributes=True)
class ClassBasicDTO(BaseModel):
@@ -108,8 +113,8 @@ class ClassBasicDTO(BaseModel):
client_id: int
class_code: str
description_spanish: Optional[str] = None
description_english: Optional[str] = None
description_es: Optional[str] = None
description_en: Optional[str] = None
material_key: Optional[str] = None
fraction: Optional[str] = None

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

@@ -4,7 +4,7 @@ Modelos ORM para gestión de clases SCAII y SCAF
from typing import TYPE_CHECKING, Optional
from api.v1.common.base_models import TenantScopedMixin
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
from sqlalchemy import (
ForeignKey,
@@ -22,7 +22,7 @@ if TYPE_CHECKING:
from api.v1.modules.public.reference_data.material_types.models import MaterialType
class Class(Base, TenantScopedMixin):
class Class(Base, TenantScopedMixin, TimestampMixin):
"""
Modelo para la tabla GClases - Información de clases en sistemas SCAII y SCAF
"""

296
backend/api/v1/modules/a76/classes/routes.py
View File

@@ -2,285 +2,23 @@
Endpoints API para gestión de clases SCAII y SCAF
"""
from typing import List, Optional
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from core.database import get_core_db
from core.security import get_current_user
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from .dto import (
ClassBasicDTO,
ClassCreateDTO,
ClassListDTO,
ClassResponseDTO,
ClassSearchDTO,
ClassUpdateDTO,
)
from .dto import ClassCreateDTO, ClassResponseDTO, ClassUpdateDTO
from .service import ClassService
router = APIRouter(prefix="/classes")
@router.get("/", response_model=ClassListDTO)
async def list_classes(
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(
100, ge=1, le=1000, description="Maximum number of records to return"
),
client_id: Optional[int] = Query(None, description="Filter by client key"),
class_code: Optional[str] = Query(None, description="Search by class code"),
description: Optional[str] = Query(None, description="Search in descriptions"),
material_key: Optional[str] = Query(None, description="Filter by material key"),
fraction: Optional[str] = Query(None, description="Filter by tariff fraction"),
physical_review: Optional[int] = Query(
None, description="Filter by physical review indicator"
),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
List classes with optional filters and pagination
"""
# Validate access to the tenant and company
tenant_id = current_user.get("tenant_id")
company_id = current_user.get("company_id")
if not tenant_id or not company_id:
raise HTTPException(
status_code=403, detail="Access denied: Tenant or Company not found"
)
service = ClassService(db)
search_params = ClassSearchDTO(
client_id=client_id,
class_code=class_code,
description=description,
material_key=material_key,
fraction=fraction,
physical_review=physical_review,
)
return service.list_classes(skip, limit, search_params)
@router.get("/client/{client_id}", response_model=List[ClassBasicDTO])
async def get_classes_by_client(
client_id: int,
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get all classes for a specific client
"""
# Validate access to the tenant and company
tenant_id = current_user.get("tenant_id")
company_id = current_user.get("company_id")
if not tenant_id or not company_id:
raise HTTPException(
status_code=403, detail="Access denied: Tenant or Company not found"
)
service = ClassService(db)
return service.search_by_client(client_id, skip, limit)
@router.get("/search/fraction/{fraction}", response_model=List[ClassBasicDTO])
async def search_by_fraction(
fraction: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Search classes by tariff fraction
"""
service = ClassService(db)
return service.search_by_fraction(fraction)
@router.get("/search/material/{material_key}", response_model=List[ClassBasicDTO])
async def search_by_material(
material_key: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Search classes by material key
"""
service = ClassService(db)
return service.search_by_material(material_key)
@router.get(
"/search/unit-measure/{unit_of_measure}", response_model=List[ClassBasicDTO]
)
async def get_classes_by_unit_measure(
unit_of_measure: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get classes by unit of measure
"""
service = ClassService(db)
return service.get_classes_by_unit_measure(unit_of_measure)
@router.get(
"/search/physical-review/{physical_review}", response_model=List[ClassBasicDTO]
)
async def get_classes_by_physical_review(
physical_review: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get classes by physical review indicator
"""
service = ClassService(db)
return service.get_classes_by_physical_review(physical_review)
@router.get("/statistics", response_model=dict)
async def get_classes_statistics(
db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)
):
"""
Get basic classes statistics
"""
service = ClassService(db)
return service.get_classes_statistics()
@router.get("/{client_id}/{class_code}", response_model=ClassResponseDTO)
async def get_class(
client_id: int,
class_code: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get class by composite key (client_id + class_code)
"""
service = ClassService(db)
class_obj = service.get_class(client_id, class_code)
if not class_obj:
raise HTTPException(
status_code=404,
detail=f"Class with client_id '{client_id}' and class_code '{class_code}' not found",
)
return class_obj
@router.post("/", response_model=ClassResponseDTO, status_code=status.HTTP_201_CREATED)
async def create_class(
class_data: ClassCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Create a new class in the system
"""
service = ClassService(db)
return service.create_class(class_data)
@router.put("/{client_id}/{class_code}", response_model=ClassResponseDTO)
async def update_class(
client_id: int,
class_code: str,
class_data: ClassUpdateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Update class information
"""
service = ClassService(db)
class_obj = service.update_class(client_id, class_code, class_data)
if not class_obj:
raise HTTPException(
status_code=404,
detail=f"Class with client_id '{client_id}' and class_code '{class_code}' not found",
)
return class_obj
@router.delete("/{client_id}/{class_code}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_class(
client_id: int,
class_code: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Delete class from the system
Note: This will completely remove the class from the system.
"""
service = ClassService(db)
if not service.delete_class(client_id, class_code):
raise HTTPException(
status_code=404,
detail=f"Class with client_id '{client_id}' and class_code '{class_code}' not found",
)
# Endpoints específicos para información detallada
@router.get("/{client_id}/{class_code}/basic", response_model=ClassBasicDTO)
async def get_class_basic_info(
client_id: int,
class_code: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get basic information for a class
"""
service = ClassService(db)
class_obj = service.get_class(client_id, class_code)
if not class_obj:
raise HTTPException(
status_code=404,
detail=f"Class with client_id '{client_id}' and class_code '{class_code}' not found",
)
return ClassBasicDTO(
client_id=class_obj.client_id,
class_code=class_obj.class_code,
description_spanish=class_obj.description_spanish,
description_english=class_obj.description_english,
material_key=class_obj.material_key,
fraction=class_obj.fraction,
)
@router.get("/{client_id}/{class_code}/tariff", response_model=dict)
async def get_class_tariff_info(
client_id: int,
class_code: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Get tariff information for a class (fractions, IVA exempt, etc.)
"""
service = ClassService(db)
class_obj = service.get_class(client_id, class_code)
if not class_obj:
raise HTTPException(
status_code=404,
detail=f"Class with client_id '{client_id}' and class_code '{class_code}' not found",
)
return {
"client_id": class_obj.client_id,
"class_code": class_obj.class_code,
"fraction": class_obj.fraction,
"us_fraction": class_obj.us_fraction,
"iva_exempt_fraction": class_obj.iva_exempt_fraction,
"sub_key": class_obj.sub_key,
"physical_review": class_obj.physical_review,
}
# Create router with generic CRUD routes
router = TenantCRUDRoutes(
service=ClassService,
create_schema=ClassCreateDTO,
update_schema=ClassUpdateDTO,
response_schema=ClassResponseDTO,
prefix="/classes",
tags=["a76 / classes"],
resource_name="Class",
id_name="class_id",
enable_list=True,
enable_filters=True,
default_page_size=50,
max_page_size=100,
).router

156
backend/api/v1/modules/a76/classes/service.py
View File

@@ -3,7 +3,7 @@ Capa de servicio para lógica de negocio de clases SCAII y SCAF
"""
import logging
from typing import List, Optional
from typing import Any, Dict, List, Optional
from fastapi import HTTPException
from sqlalchemy import and_, or_
@@ -26,6 +26,160 @@ logger = logging.getLogger(__name__)
class ClassService:
"""Servicio para gestión de clases SCAII y SCAF"""
@staticmethod
def get_all(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 100,
filters: Optional[Dict[str, Any]] = None,
) -> tuple[List[Class], int]:
"""
Get all classes for a tenant with pagination and filters
"""
query = db.query(Class).filter(
Class.tenant_id == tenant_id, Class.company_id == company_id
)
if filters:
if filters.get("client_id"):
query = query.filter(Class.client_id == filters["client_id"])
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']}%"))
if filters.get("physical_review") is not None:
query = query.filter(
Class.physical_review == filters["physical_review"]
)
total = query.count()
items = query.offset(skip).limit(limit).all()
return items, total
@staticmethod
def get_by_id(
db: Session, class_id: int, tenant_id: int, company_id: int
) -> Optional[Class]:
"""Get a class by ID"""
return (
db.query(Class)
.filter(
Class.id == class_id,
Class.tenant_id == tenant_id,
Class.company_id == company_id,
)
.first()
)
@staticmethod
def create(
db: Session, class_data: ClassCreateDTO, tenant_id: int, company_id: int
) -> Class:
"""Create a new class"""
from fastapi import HTTPException
from sqlalchemy.exc import IntegrityError
data_dict = class_data.model_dump()
# Check if class_code already exists for this tenant and company
existing = db.query(Class).filter(
Class.tenant_id == tenant_id,
Class.company_id == company_id,
Class.class_code == data_dict["class_code"]
).first()
if existing:
raise HTTPException(
status_code=400,
detail=f"Class with code '{data_dict['class_code']}' already exists for this tenant and company"
)
# Validate material_key exists if provided
if data_dict.get("material_key"):
from api.v1.modules.public.reference_data.material_types.models import MaterialType
material_exists = db.query(MaterialType).filter(
MaterialType.key == data_dict["material_key"]
).first()
if not material_exists:
# Set to None if material_key doesn't exist
data_dict["material_key"] = None
class_obj = Class(**data_dict)
class_obj.tenant_id = tenant_id
class_obj.company_id = company_id
try:
db.add(class_obj)
db.commit()
db.refresh(class_obj)
return class_obj
except IntegrityError as e:
db.rollback()
raise HTTPException(
status_code=400,
detail=f"Failed to create class: {str(e.orig)}"
)
@staticmethod
def update(
db: Session,
class_id: int,
tenant_id: int,
class_data: ClassUpdateDTO,
company_id: int,
) -> Optional[Class]:
"""Update a class"""
class_obj = ClassService.get_by_id(db, class_id, tenant_id, company_id)
if not class_obj:
return None
update_data = class_data.model_dump(exclude_unset=True)
# Validate material_key exists if provided
if "material_key" in update_data and update_data["material_key"]:
from api.v1.modules.public.reference_data.material_types.models import MaterialType
material_exists = db.query(MaterialType).filter(
MaterialType.key == update_data["material_key"]
).first()
if not material_exists:
# Set to None if material_key doesn't exist
update_data["material_key"] = None
for field, value in update_data.items():
setattr(class_obj, field, value)
db.commit()
db.refresh(class_obj)
return class_obj
@staticmethod
def delete(db: Session, class_id: int, tenant_id: int, company_id: int) -> bool:
"""Delete a class"""
class_obj = ClassService.get_by_id(db, class_id, tenant_id, company_id)
if not class_obj:
return False
db.delete(class_obj)
db.commit()
return True
def __init__(self, db: Session):
self.db = db