acazares
b68c4316ff
- Rearranged imports in multiple files for consistency and clarity. - Updated logging middleware to exclude specific paths from logging. - Enhanced security module by cleaning up token handling and improving tenant validation. - Added tenant and company scoped mixins for better database model management. - Implemented generic CRUD routes for tenant-scoped resources. - Improved error handling and response management in API routes. - Cleaned up login and logout processes to ensure proper session management. - Introduced mechanisms to clear local storage and cookies on tenant change. - Enhanced company store to detect tenant changes and clear data accordingly. - Added new DTO mixins for currency and value affect flags.
310 lines
10 KiB
Python
310 lines
10 KiB
Python
"""
|
|
Capa de servicio para lógica de negocio de partes/componentes
|
|
"""
|
|
|
|
import logging
|
|
from typing import List, Optional
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import and_, func, or_
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .dto import PartCreateDTO, PartUpdateDTO
|
|
from .models import Part
|
|
|
|
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"
|
|
)
|