feat: Convertir módulo de clases para usar TenantCRUDRoutes
Cambios en backend: - Actualizado ClassResponseDTO con campos de tenant y timestamps - Agregados métodos estáticos en servicio (get_all, get_by_id, create, update, delete) - Reemplazados endpoints de rutas con inicialización de TenantCRUDRoutes - Corregida ruta de importación del modelo Company en security.py - Actualizados nombres de campos en DTO para coincidir con modelo (description_es/en) - Agregada validación para constraint de foreign key de material_key - Agregada validación de class_code duplicado en método create - Agregado TimestampMixin al modelo Class - Creada migración de Alembic para agregar columnas timestamp a tabla classes - Corregido orden de parámetros en firma del método update Cambios en frontend: - Actualizados componentes de UI para módulo de clases
This commit is contained in:
KevinMrkz3221
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user