acazares
e0e38526b0
- Updated PermissionRuleOct model to include company_id and modified unique constraint. - Updated Seal model to include company_id and modified unique constraint. - Added TYPE_CHECKING imports in reference data models for better type hinting. - Created new models for QClasses and SClasses in the a24 module. - Enhanced init_first_time.sh script for comprehensive system initialization, including Keycloak and PostgreSQL setup. - Updated documentation to reflect changes in API endpoints and database relationships.
276 lines
9.9 KiB
Python
276 lines
9.9 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 Part
|
|
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) -> Part:
|
|
"""
|
|
Crear una nueva parte
|
|
"""
|
|
try:
|
|
db_part = Part(**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_id 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_id: int, part_number: str) -> Optional[Part]:
|
|
"""
|
|
Obtener una parte por clave de cliente y número de parte
|
|
"""
|
|
try:
|
|
return db.query(Part).filter(
|
|
and_(
|
|
Part.client_id == client_id,
|
|
Part.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_id: Optional[int] = None,
|
|
fraction: Optional[str] = None,
|
|
country_of_origin: Optional[str] = None
|
|
) -> tuple[List[Part], int]:
|
|
"""
|
|
Obtener partes con paginación y filtros
|
|
"""
|
|
try:
|
|
query = db.query(Part)
|
|
|
|
# Aplicar filtros
|
|
if search:
|
|
query = query.filter(or_(
|
|
Part.description_spanish.ilike(f"%{search}%"),
|
|
Part.description_english.ilike(f"%{search}%"),
|
|
Part.part_number.ilike(f"%{search}%")
|
|
))
|
|
|
|
if client_id is not None:
|
|
query = query.filter(Part.client_id == client_id)
|
|
|
|
if fraction:
|
|
query = query.filter(Part.fraction == fraction)
|
|
|
|
if country_of_origin:
|
|
query = query.filter(Part.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_id: int) -> List[Part]:
|
|
"""
|
|
Obtener todas las partes de un cliente específico
|
|
"""
|
|
try:
|
|
return db.query(Part).filter(Part.client_id == client_id).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[Part]:
|
|
"""
|
|
Buscar partes por fracción arancelaria
|
|
"""
|
|
try:
|
|
return db.query(Part).filter(
|
|
or_(
|
|
Part.fraction.ilike(f"%{fraction}%"),
|
|
Part.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[Part]:
|
|
"""
|
|
Buscar partes por proveedor
|
|
"""
|
|
try:
|
|
return db.query(Part).filter(Part.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[Part]:
|
|
"""
|
|
Buscar partes por país de origen
|
|
"""
|
|
try:
|
|
return db.query(Part).filter(Part.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_id: int, part_number: str, part_data: PartUpdateDTO) -> Optional[Part]:
|
|
"""
|
|
Actualizar una parte existente
|
|
"""
|
|
try:
|
|
db_part = PartService.get_part(db, client_id, 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_id: int, part_number: str) -> bool:
|
|
"""
|
|
Eliminar una parte
|
|
"""
|
|
try:
|
|
db_part = PartService.get_part(db, client_id, 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_id: int, part_number: str) -> Optional[Part]:
|
|
"""
|
|
Cambiar el estado habilitado/deshabilitado de una parte
|
|
"""
|
|
try:
|
|
db_part = PartService.get_part(db, client_id, 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(Part).count()
|
|
|
|
# Partes por cliente
|
|
parts_by_client = db.query(
|
|
Part.client_id,
|
|
func.count(Part.part_number).label('count')
|
|
).group_by(Part.client_id).all()
|
|
|
|
# Partes por país de origen
|
|
parts_by_country = db.query(
|
|
Part.country_of_origin,
|
|
func.count(Part.part_number).label('count')
|
|
).filter(Part.country_of_origin.isnot(None))\
|
|
.group_by(Part.country_of_origin).all()
|
|
|
|
# Partes habilitadas vs deshabilitadas
|
|
enabled_parts = db.query(Part).filter(Part.enabled_disabled == 1).count()
|
|
disabled_parts = db.query(Part).filter(Part.enabled_disabled == 0).count()
|
|
|
|
return {
|
|
"total_parts": total_parts,
|
|
"enabled_parts": enabled_parts,
|
|
"disabled_parts": disabled_parts,
|
|
"parts_by_client": [{"client_id": 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_id: int, part_number: str) -> Optional[dict]:
|
|
"""
|
|
Obtener información regulatoria específica de una parte
|
|
"""
|
|
try:
|
|
db_part = PartService.get_part(db, client_id, part_number)
|
|
if not db_part:
|
|
return None
|
|
|
|
return {
|
|
"client_id": db_part.client_id,
|
|
"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")
|
|
|