706 lines
25 KiB
Python
706 lines
25 KiB
Python
"""
|
|
Capa de servicio para lógica de negocio de clases SCAII y SCAF
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import and_, or_
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .dto import (
|
|
ClassBasicDTO,
|
|
ClassCreateDTO,
|
|
ClassCreateDTOFA,
|
|
ClassListDTO,
|
|
ClassResponseDTO,
|
|
ClassResponseDTOFA,
|
|
ClassSearchDTO,
|
|
ClassUpdateDTO,
|
|
)
|
|
from .models import Class
|
|
|
|
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: Optional[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)
|
|
|
|
if company_id is not None:
|
|
query = query.filter(Class.company_id == company_id)
|
|
|
|
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']}%"))
|
|
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_all_with_fa_data(
|
|
db: Session,
|
|
tenant_id: int,
|
|
company_id: Optional[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)
|
|
)
|
|
|
|
if company_id is not None:
|
|
query = query.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,
|
|
"import_tariff_code": fa_class.import_tariff_code if fa_class else None,
|
|
"import_tariff_type": fa_class.import_tariff_type if fa_class else None,
|
|
"export_tariff_code": fa_class.export_tariff_code if fa_class else None,
|
|
"export_tariff_type": fa_class.export_tariff_type 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,
|
|
"eccn_code": fa_class.eccn_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
|
|
) -> 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" El código de clase '{data_dict['class_code']}' ya existe. Por favor use un código diferente."
|
|
)
|
|
|
|
# Validate material_key exists (now required)
|
|
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:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Material type '{data_dict['material_key']}' does not exist"
|
|
)
|
|
|
|
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:
|
|
logger.warning(f"Class {class_id} not found for tenant {tenant_id}, company {company_id}")
|
|
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
|
|
|
|
# Validate class_code is unique if being changed
|
|
if "class_code" in update_data and update_data["class_code"]:
|
|
new_code = update_data["class_code"]
|
|
# Check if another class with this code exists (excluding current class)
|
|
# The unique constraint is on (tenant_id, company_id, class_code)
|
|
existing_class = db.query(Class).filter(
|
|
Class.class_code == new_code,
|
|
Class.tenant_id == tenant_id,
|
|
Class.company_id == company_id,
|
|
Class.id != class_id # Exclude current class
|
|
).first()
|
|
|
|
if existing_class:
|
|
logger.warning(f"Duplicate class_code found: {existing_class.id}")
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"El código '{new_code}' ya está en uso para este cliente. Por favor ingrese un código diferente."
|
|
)
|
|
|
|
for field, value in update_data.items():
|
|
setattr(class_obj, field, value)
|
|
|
|
try:
|
|
db.commit()
|
|
db.refresh(class_obj)
|
|
return class_obj
|
|
except IntegrityError as e:
|
|
db.rollback()
|
|
error_msg = str(e.orig)
|
|
logger.error(f"IntegrityError updating class {class_id}: {error_msg}")
|
|
|
|
# Check if it's a duplicate class_code error
|
|
if "already exists" in error_msg.lower() or "duplicate" in error_msg.lower():
|
|
# Extract the code from update_data if it was changed
|
|
code = update_data.get("class_code", class_obj.class_code)
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"El código '{code}' ya está en uso. Por favor ingrese un código diferente."
|
|
)
|
|
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Error al actualizar la clase: {error_msg}"
|
|
)
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Unexpected error updating class {class_id}: {type(e).__name__}: {str(e)}")
|
|
raise
|
|
|
|
@staticmethod
|
|
def delete(db: Session, class_id: int, tenant_id: int, company_id: int) -> bool:
|
|
"""Delete a class (and its FA extension if exists)"""
|
|
from api.v1.modules.a24.fa.fa_classes.models import QClasses
|
|
|
|
class_obj = ClassService.get_by_id(db, class_id, tenant_id, company_id)
|
|
if not class_obj:
|
|
return False
|
|
|
|
# Delete FA extension first (if exists) to avoid FK constraint violation
|
|
fa_extension = db.query(QClasses).filter(
|
|
QClasses.class_id == class_id,
|
|
QClasses.tenant_id == tenant_id
|
|
).first()
|
|
|
|
if fa_extension:
|
|
db.delete(fa_extension)
|
|
|
|
# Now delete the base class
|
|
db.delete(class_obj)
|
|
db.commit()
|
|
return True
|
|
|
|
@staticmethod
|
|
def create_fa_class(
|
|
db: Session, class_data: ClassCreateDTOFA, tenant_id: int, company_id: int
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Create a fixed asset class (both a76.classes and a24.fa_classes)
|
|
Returns a dict with both records combined
|
|
"""
|
|
|
|
from api.v1.modules.a24.fa.fa_classes.models import QClasses
|
|
|
|
# Extract base class fields
|
|
base_fields = {
|
|
"class_code", "description_es", "description_en",
|
|
"material_key", "unit_of_measure", "fraction", "us_fraction",
|
|
"sub_key", "physical_review", "iva_exempt_fraction"
|
|
}
|
|
base_data = {k: v for k, v in class_data.model_dump().items() if k in base_fields}
|
|
|
|
# Extract FA-specific fields
|
|
fa_fields = {
|
|
"import_tariff_code", "import_tariff_type", "export_tariff_code",
|
|
"export_tariff_type", "depreciation_rate", "fda_code", "eccn_code",
|
|
"class_enabled"
|
|
}
|
|
fa_data = {k: v for k, v in class_data.model_dump().items() if k in fa_fields}
|
|
|
|
try:
|
|
# 1. Create base class
|
|
base_dto = ClassCreateDTO(**base_data)
|
|
base_class = ClassService.create(db, base_dto, tenant_id, company_id)
|
|
|
|
# 2. Create FA extension
|
|
fa_obj = QClasses(**fa_data)
|
|
fa_obj.class_id = base_class.id
|
|
fa_obj.tenant_id = tenant_id
|
|
fa_obj.company_id = company_id
|
|
|
|
db.add(fa_obj)
|
|
db.commit()
|
|
db.refresh(fa_obj)
|
|
|
|
# 3. Combine response - build dict manually to avoid SQLAlchemy internals
|
|
combined_response = {
|
|
# 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
|
|
"fa_id": fa_obj.id,
|
|
"import_tariff_code": fa_obj.import_tariff_code,
|
|
"import_tariff_type": fa_obj.import_tariff_type,
|
|
"export_tariff_code": fa_obj.export_tariff_code,
|
|
"export_tariff_type": fa_obj.export_tariff_type,
|
|
"depreciation_rate": fa_obj.depreciation_rate,
|
|
"fda_code": fa_obj.fda_code,
|
|
"eccn_code": fa_obj.eccn_code,
|
|
"class_enabled": fa_obj.class_enabled,
|
|
}
|
|
|
|
return combined_response
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
# If FA creation fails, rollback base class too
|
|
if 'base_class' in locals():
|
|
try:
|
|
db.delete(base_class)
|
|
db.commit()
|
|
except:
|
|
pass
|
|
|
|
# Extract and improve error message
|
|
error_msg = str(e)
|
|
if "already exists" in error_msg.lower() or "duplicad" in error_msg.lower():
|
|
# Extract code from error if possible
|
|
code = class_data.class_code
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"El código '{code}' ya está en uso. Por favor ingrese un código diferente."
|
|
)
|
|
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Error al crear clase de activo fijo: {error_msg}"
|
|
)
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def create_class(self, class_data: ClassCreateDTO) -> ClassResponseDTO:
|
|
"""
|
|
Crea una nueva clase en el sistema
|
|
|
|
Args:
|
|
class_data: Datos de la clase a crear
|
|
|
|
Returns:
|
|
ClassResponseDTO con información de la clase creada
|
|
|
|
Raises:
|
|
HTTPException: Si la clase ya existe o error en la creación
|
|
"""
|
|
try:
|
|
# Verificar que no exista la clase
|
|
existing = (
|
|
self.db.query(Class)
|
|
.filter(
|
|
and_(
|
|
Class.class_code == class_data.class_code,
|
|
)
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if existing:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Class with class_code '{class_data.class_code}' already exists",
|
|
)
|
|
|
|
# Crear clase
|
|
db_class = Class(
|
|
class_code=class_data.class_code,
|
|
description_spanish=class_data.description_spanish,
|
|
description_english=class_data.description_english,
|
|
material_key=class_data.material_key,
|
|
unit_of_measure=class_data.unit_of_measure,
|
|
fraction=class_data.fraction,
|
|
us_fraction=class_data.us_fraction,
|
|
sub_key=class_data.sub_key,
|
|
physical_review=class_data.physical_review,
|
|
iva_exempt_fraction=class_data.iva_exempt_fraction,
|
|
)
|
|
|
|
self.db.add(db_class)
|
|
self.db.commit()
|
|
self.db.refresh(db_class)
|
|
|
|
return ClassResponseDTO.model_validate(db_class)
|
|
|
|
except IntegrityError as e:
|
|
self.db.rollback()
|
|
logger.error(f"IntegrityError creating class: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Class with this class_code already exists",
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Error creating class: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Error creating class")
|
|
|
|
def get_class(self, class_code: str) -> Optional[ClassResponseDTO]:
|
|
"""
|
|
Obtiene una clase por clave compuesta
|
|
|
|
Args:
|
|
class_code: Código de clase
|
|
|
|
Returns:
|
|
ClassResponseDTO o None si no existe
|
|
"""
|
|
class_obj = (
|
|
self.db.query(Class)
|
|
.filter(and_(Class.class_code == class_code))
|
|
.first()
|
|
)
|
|
|
|
if not class_obj:
|
|
return None
|
|
return ClassResponseDTO.model_validate(class_obj)
|
|
|
|
def list_classes(
|
|
self,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
search_params: Optional[ClassSearchDTO] = None,
|
|
) -> ClassListDTO:
|
|
"""
|
|
Lista clases con filtros
|
|
|
|
Args:
|
|
skip: Número de registros a omitir
|
|
limit: Número máximo de registros a retornar
|
|
search_params: Parámetros de búsqueda
|
|
|
|
Returns:
|
|
ClassListDTO con la lista paginada
|
|
"""
|
|
query = self.db.query(Class)
|
|
|
|
# Aplicar filtros si se proporcionan
|
|
if search_params:
|
|
if search_params.class_code:
|
|
query = query.filter(
|
|
Class.class_code.ilike(f"%{search_params.class_code}%")
|
|
)
|
|
|
|
if search_params.description:
|
|
description_pattern = f"%{search_params.description}%"
|
|
query = query.filter(
|
|
or_(
|
|
Class.description_spanish.ilike(description_pattern),
|
|
Class.description_english.ilike(description_pattern),
|
|
)
|
|
)
|
|
|
|
if search_params.material_key:
|
|
query = query.filter(
|
|
Class.material_key.ilike(f"%{search_params.material_key}%")
|
|
)
|
|
|
|
if search_params.fraction:
|
|
query = query.filter(
|
|
Class.fraction.ilike(f"%{search_params.fraction}%")
|
|
)
|
|
|
|
if search_params.physical_review is not None:
|
|
query = query.filter(
|
|
Class.physical_review == search_params.physical_review
|
|
)
|
|
|
|
# Contar total
|
|
total = query.count()
|
|
|
|
# Aplicar paginación
|
|
classes = query.offset(skip).limit(limit).all()
|
|
|
|
# Convertir a DTOs básicos
|
|
class_dtos = [ClassBasicDTO.model_validate(class_obj) for class_obj in classes]
|
|
|
|
return ClassListDTO(
|
|
classes=class_dtos,
|
|
total=total,
|
|
page=(skip // limit) + 1 if limit > 0 else 1,
|
|
size=len(class_dtos),
|
|
)
|
|
|
|
def update_class(
|
|
self, class_code: str, class_data: ClassUpdateDTO
|
|
) -> Optional[ClassResponseDTO]:
|
|
"""
|
|
Actualiza una clase
|
|
|
|
Args:
|
|
class_code: Código de clase
|
|
class_data: Datos a actualizar
|
|
|
|
Returns:
|
|
ClassResponseDTO actualizado o None si no existe
|
|
"""
|
|
class_obj = (
|
|
self.db.query(Class)
|
|
.filter(and_(Class.class_code == class_code))
|
|
.first()
|
|
)
|
|
|
|
if not class_obj:
|
|
return None
|
|
|
|
try:
|
|
# Actualizar solo campos proporcionados
|
|
update_data = class_data.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(class_obj, field, value)
|
|
|
|
self.db.commit()
|
|
self.db.refresh(class_obj)
|
|
|
|
return ClassResponseDTO.model_validate(class_obj)
|
|
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Error updating class {class_code}: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Error updating class")
|
|
|
|
def delete_class(self, class_code: str) -> bool:
|
|
"""
|
|
Elimina una clase
|
|
|
|
Args:
|
|
class_code: Código de clase
|
|
|
|
Returns:
|
|
True si se eliminó, False si no existe
|
|
"""
|
|
class_obj = (
|
|
self.db.query(Class)
|
|
.filter(and_(Class.class_code == class_code))
|
|
.first()
|
|
)
|
|
|
|
if not class_obj:
|
|
return False
|
|
|
|
try:
|
|
self.db.delete(class_obj)
|
|
self.db.commit()
|
|
return True
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Error deleting class {class_code}: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Error deleting class")
|
|
|
|
def search_by_fraction(self, fraction: str) -> List[ClassBasicDTO]:
|
|
"""Busca clases por fracción arancelaria"""
|
|
classes = (
|
|
self.db.query(Class).filter(Class.fraction.ilike(f"%{fraction}%")).all()
|
|
)
|
|
return [ClassBasicDTO.model_validate(class_obj) for class_obj in classes]
|
|
|
|
def search_by_material(self, material_key: str) -> List[ClassBasicDTO]:
|
|
"""Busca clases por clave de material"""
|
|
classes = (
|
|
self.db.query(Class)
|
|
.filter(Class.material_key.ilike(f"%{material_key}%"))
|
|
.all()
|
|
)
|
|
return [ClassBasicDTO.model_validate(class_obj) for class_obj in classes]
|
|
|
|
def get_classes_by_physical_review(
|
|
self, physical_review: int
|
|
) -> List[ClassBasicDTO]:
|
|
"""Obtiene clases por indicador de revisión física"""
|
|
classes = (
|
|
self.db.query(Class).filter(Class.physical_review == physical_review).all()
|
|
)
|
|
return [ClassBasicDTO.model_validate(class_obj) for class_obj in classes]
|
|
|
|
def get_classes_statistics(self) -> dict:
|
|
"""Obtiene estadísticas básicas de clases"""
|
|
total_classes = self.db.query(Class).count()
|
|
|
|
# Contar por revisión física
|
|
physical_review_stats = {}
|
|
for i in range(3): # Asumiendo valores 0, 1, 2
|
|
count = self.db.query(Class).filter(Class.physical_review == i).count()
|
|
physical_review_stats[f"physical_review_{i}"] = count
|
|
|
|
# Contar clases con fracciones
|
|
with_fraction = self.db.query(Class).filter(Class.fraction.isnot(None)).count()
|
|
with_us_fraction = (
|
|
self.db.query(Class).filter(Class.us_fraction.isnot(None)).count()
|
|
)
|
|
|
|
return {
|
|
"total_classes": total_classes,
|
|
"classes_with_fraction": with_fraction,
|
|
"classes_with_us_fraction": with_us_fraction,
|
|
**physical_review_stats,
|
|
}
|
|
|
|
def get_classes_by_unit_measure(self, unit_of_measure: str) -> List[ClassBasicDTO]:
|
|
"""Obtiene clases por unidad de medida"""
|
|
classes = (
|
|
self.db.query(Class).filter(Class.unit_of_measure == unit_of_measure).all()
|
|
)
|
|
return [ClassBasicDTO.model_validate(class_obj) for class_obj in classes]
|