Kevin Rosales
c1dee97092
## Cambios Realizados ### 1. **Modelos** - Se actualizaron los modelos para incluir el esquema en las tablas: - - - - - Ajustes en relaciones y claves foráneas para garantizar consistencia con el esquema . - Se añadieron anotaciones de tipo y mejoras en la documentación de los modelos. ### 2. **Servicios** - Implementación de lógica de negocio en para el módulo : - Creación, actualización, eliminación y búsqueda de partes. - Métodos para estadísticas y manejo de estados habilitado/deshabilitado. ### 3. **Migraciones** - Creación de nuevas migraciones de Alembic para las tablas: - - - - - Tablas relacionadas con direcciones y programas de clientes/proveedores. ### 4. **Documentación** - Actualización de : - Detalles de las nuevas funcionalidades implementadas. - Endpoints REST API agregados para los módulos. - Relaciones principales entre tablas. - Actualización de : - Cambios realizados en los modelos para usar el esquema . - Beneficios de la separación de esquemas. - Próximos pasos para completar la integración. ## Próximos Pasos 1. Verificar las migraciones generadas y aplicarlas en el entorno de desarrollo. 2. Implementar pruebas unitarias para los nuevos servicios y modelos. --- *Documento generado automáticamente el 5 de noviembre de 2025.*
276 lines
10 KiB
Python
276 lines
10 KiB
Python
"""
|
|
Capa de servicio para lógica de negocio de partes/componentes
|
|
"""
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy import or_, and_, func
|
|
from fastapi import HTTPException
|
|
from typing import List, Optional
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from .models import GPart
|
|
from .dto import PartCreateDTO, PartUpdateDTO
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PartService:
|
|
"""
|
|
Servicio para gestión de partes/componentes
|
|
"""
|
|
|
|
@staticmethod
|
|
def create_part(db: Session, part_data: PartCreateDTO) -> GPart:
|
|
"""
|
|
Crear una nueva parte
|
|
"""
|
|
try:
|
|
db_part = GPart(**part_data.model_dump())
|
|
db.add(db_part)
|
|
db.commit()
|
|
db.refresh(db_part)
|
|
return db_part
|
|
except IntegrityError as e:
|
|
db.rollback()
|
|
logger.error(f"Error creating part: {e}")
|
|
raise HTTPException(status_code=400, detail="Part with this client_key and part_number already exists")
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Unexpected error creating part: {e}")
|
|
raise HTTPException(status_code=500, detail="Error creating part")
|
|
|
|
@staticmethod
|
|
def get_part(db: Session, client_key: int, part_number: str) -> Optional[GPart]:
|
|
"""
|
|
Obtener una parte por clave de cliente y número de parte
|
|
"""
|
|
try:
|
|
return db.query(GPart).filter(
|
|
and_(
|
|
GPart.client_key == client_key,
|
|
GPart.part_number == part_number
|
|
)
|
|
).first()
|
|
except Exception as e:
|
|
logger.error(f"Error getting part: {e}")
|
|
raise HTTPException(status_code=500, detail="Error retrieving part")
|
|
|
|
@staticmethod
|
|
def get_parts_paginated(
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
search: Optional[str] = None,
|
|
client_key: Optional[int] = None,
|
|
fraction: Optional[str] = None,
|
|
country_of_origin: Optional[str] = None
|
|
) -> tuple[List[GPart], int]:
|
|
"""
|
|
Obtener partes con paginación y filtros
|
|
"""
|
|
try:
|
|
query = db.query(GPart)
|
|
|
|
# Aplicar filtros
|
|
if search:
|
|
query = query.filter(or_(
|
|
GPart.description_spanish.ilike(f"%{search}%"),
|
|
GPart.description_english.ilike(f"%{search}%"),
|
|
GPart.part_number.ilike(f"%{search}%")
|
|
))
|
|
|
|
if client_key is not None:
|
|
query = query.filter(GPart.client_key == client_key)
|
|
|
|
if fraction:
|
|
query = query.filter(GPart.fraction == fraction)
|
|
|
|
if country_of_origin:
|
|
query = query.filter(GPart.country_of_origin == country_of_origin)
|
|
|
|
# Contar total
|
|
total = query.count()
|
|
|
|
# Aplicar paginación
|
|
parts = query.offset(skip).limit(limit).all()
|
|
|
|
return parts, total
|
|
except Exception as e:
|
|
logger.error(f"Error getting paginated parts: {e}")
|
|
raise HTTPException(status_code=500, detail="Error retrieving parts")
|
|
|
|
@staticmethod
|
|
def get_parts_by_client(db: Session, client_key: int) -> List[GPart]:
|
|
"""
|
|
Obtener todas las partes de un cliente específico
|
|
"""
|
|
try:
|
|
return db.query(GPart).filter(GPart.client_key == client_key).all()
|
|
except Exception as e:
|
|
logger.error(f"Error getting parts by client: {e}")
|
|
raise HTTPException(status_code=500, detail="Error retrieving client parts")
|
|
|
|
@staticmethod
|
|
def search_parts_by_fraction(db: Session, fraction: str) -> List[GPart]:
|
|
"""
|
|
Buscar partes por fracción arancelaria
|
|
"""
|
|
try:
|
|
return db.query(GPart).filter(
|
|
or_(
|
|
GPart.fraction.ilike(f"%{fraction}%"),
|
|
GPart.us_fraction.ilike(f"%{fraction}%")
|
|
)
|
|
).all()
|
|
except Exception as e:
|
|
logger.error(f"Error searching parts by fraction: {e}")
|
|
raise HTTPException(status_code=500, detail="Error searching parts by fraction")
|
|
|
|
@staticmethod
|
|
def search_parts_by_supplier(db: Session, supplier: str) -> List[GPart]:
|
|
"""
|
|
Buscar partes por proveedor
|
|
"""
|
|
try:
|
|
return db.query(GPart).filter(GPart.supplier.ilike(f"%{supplier}%")).all()
|
|
except Exception as e:
|
|
logger.error(f"Error searching parts by supplier: {e}")
|
|
raise HTTPException(status_code=500, detail="Error searching parts by supplier")
|
|
|
|
@staticmethod
|
|
def search_parts_by_country(db: Session, country_code: str) -> List[GPart]:
|
|
"""
|
|
Buscar partes por país de origen
|
|
"""
|
|
try:
|
|
return db.query(GPart).filter(GPart.country_of_origin == country_code).all()
|
|
except Exception as e:
|
|
logger.error(f"Error searching parts by country: {e}")
|
|
raise HTTPException(status_code=500, detail="Error searching parts by country")
|
|
|
|
@staticmethod
|
|
def update_part(db: Session, client_key: int, part_number: str, part_data: PartUpdateDTO) -> Optional[GPart]:
|
|
"""
|
|
Actualizar una parte existente
|
|
"""
|
|
try:
|
|
db_part = PartService.get_part(db, client_key, part_number)
|
|
if not db_part:
|
|
return None
|
|
|
|
# Actualizar campos
|
|
for field, value in part_data.model_dump(exclude_unset=True).items():
|
|
setattr(db_part, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_part)
|
|
return db_part
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error updating part: {e}")
|
|
raise HTTPException(status_code=500, detail="Error updating part")
|
|
|
|
@staticmethod
|
|
def delete_part(db: Session, client_key: int, part_number: str) -> bool:
|
|
"""
|
|
Eliminar una parte
|
|
"""
|
|
try:
|
|
db_part = PartService.get_part(db, client_key, part_number)
|
|
if not db_part:
|
|
return False
|
|
|
|
db.delete(db_part)
|
|
db.commit()
|
|
return True
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error deleting part: {e}")
|
|
raise HTTPException(status_code=500, detail="Error deleting part")
|
|
|
|
@staticmethod
|
|
def toggle_part_status(db: Session, client_key: int, part_number: str) -> Optional[GPart]:
|
|
"""
|
|
Cambiar el estado habilitado/deshabilitado de una parte
|
|
"""
|
|
try:
|
|
db_part = PartService.get_part(db, client_key, part_number)
|
|
if not db_part:
|
|
return None
|
|
|
|
# Toggle status (assuming 1 = enabled, 0 = disabled)
|
|
db_part.enabled_disabled = 1 if db_part.enabled_disabled == 0 else 0
|
|
|
|
db.commit()
|
|
db.refresh(db_part)
|
|
return db_part
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error toggling part status: {e}")
|
|
raise HTTPException(status_code=500, detail="Error toggling part status")
|
|
|
|
@staticmethod
|
|
def get_parts_statistics(db: Session) -> dict:
|
|
"""
|
|
Obtener estadísticas de partes
|
|
"""
|
|
try:
|
|
total_parts = db.query(GPart).count()
|
|
|
|
# Partes por cliente
|
|
parts_by_client = db.query(
|
|
GPart.client_key,
|
|
func.count(GPart.part_number).label('count')
|
|
).group_by(GPart.client_key).all()
|
|
|
|
# Partes por país de origen
|
|
parts_by_country = db.query(
|
|
GPart.country_of_origin,
|
|
func.count(GPart.part_number).label('count')
|
|
).filter(GPart.country_of_origin.isnot(None))\
|
|
.group_by(GPart.country_of_origin).all()
|
|
|
|
# Partes habilitadas vs deshabilitadas
|
|
enabled_parts = db.query(GPart).filter(GPart.enabled_disabled == 1).count()
|
|
disabled_parts = db.query(GPart).filter(GPart.enabled_disabled == 0).count()
|
|
|
|
return {
|
|
"total_parts": total_parts,
|
|
"enabled_parts": enabled_parts,
|
|
"disabled_parts": disabled_parts,
|
|
"parts_by_client": [{"client_key": item[0], "count": item[1]} for item in parts_by_client],
|
|
"parts_by_country": [{"country": item[0], "count": item[1]} for item in parts_by_country]
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Error getting parts statistics: {e}")
|
|
raise HTTPException(status_code=500, detail="Error retrieving parts statistics")
|
|
|
|
@staticmethod
|
|
def get_part_regulatory_info(db: Session, client_key: int, part_number: str) -> Optional[dict]:
|
|
"""
|
|
Obtener información regulatoria específica de una parte
|
|
"""
|
|
try:
|
|
db_part = PartService.get_part(db, client_key, part_number)
|
|
if not db_part:
|
|
return None
|
|
|
|
return {
|
|
"client_key": db_part.client_key,
|
|
"part_number": db_part.part_number,
|
|
"fraction": db_part.fraction,
|
|
"us_fraction": db_part.us_fraction,
|
|
"fda_key": db_part.fda_key,
|
|
"fcc_key": db_part.fcc_key,
|
|
"license_code": db_part.license_code,
|
|
"eccn": db_part.eccn,
|
|
"export_code": db_part.export_code,
|
|
"exclusion_symbol": db_part.exclusion_symbol,
|
|
"country_of_origin": db_part.country_of_origin
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Error getting part regulatory info: {e}")
|
|
raise HTTPException(status_code=500, detail="Error retrieving part regulatory information")
|
|
|