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.
208 lines
7.1 KiB
Python
208 lines
7.1 KiB
Python
"""
|
|
Capa de servicio para lógica de negocio de empresa
|
|
"""
|
|
|
|
import logging
|
|
from typing import List, Optional
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .dto import CompanyCreateDTO, CompanyResponseDTO, CompanyUpdateDTO
|
|
from .models import Company
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CompanyService:
|
|
"""Servicio para gestión de empresa"""
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def create_company(self, company_data: CompanyCreateDTO) -> CompanyResponseDTO:
|
|
"""
|
|
Crea una nueva empresa en el sistema
|
|
|
|
Args:
|
|
company_data: Datos de la empresa a crear
|
|
|
|
Returns:
|
|
CompanyResponseDTO con información de la empresa creada
|
|
|
|
Raises:
|
|
HTTPException: Si ya existe una empresa o error en la creación
|
|
"""
|
|
try:
|
|
# Verificar que no exista ya una empresa (solo puede haber una por el consecutivo único)
|
|
existing = self.db.query(Company).filter(Company.id).first()
|
|
if existing:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="A company is already registered in the system",
|
|
)
|
|
|
|
# Crear empresa
|
|
db_company = Company(
|
|
id=company_data.id,
|
|
consecutive=company_data.consecutive,
|
|
name=company_data.name,
|
|
rfc=company_data.rfc,
|
|
main_activity=company_data.main_activity,
|
|
program=company_data.program,
|
|
program_number=company_data.program_number,
|
|
prosec=company_data.prosec,
|
|
prosec_authorization=company_data.prosec_authorization,
|
|
manufacturer_id=company_data.manufacturer_id,
|
|
broker_company=company_data.broker_company,
|
|
responsible=company_data.responsible,
|
|
responsible_name=company_data.responsible_name,
|
|
responsible_last_name=company_data.responsible_last_name,
|
|
responsible_mother_last_name=company_data.responsible_mother_last_name,
|
|
responsible_rfc=company_data.responsible_rfc,
|
|
position=company_data.position,
|
|
logo=company_data.logo,
|
|
has_express_line=company_data.has_express_line,
|
|
order_format_type=company_data.order_format_type,
|
|
previous_code=company_data.previous_code,
|
|
is_service_company=company_data.is_service_company,
|
|
client_name=company_data.client_name,
|
|
subassembly_mode=company_data.subassembly_mode,
|
|
curp=company_data.curp,
|
|
inter_db_name=company_data.inter_db_name,
|
|
ctpat_svi=company_data.ctpat_svi,
|
|
trusted_exporter_number=company_data.trusted_exporter_number,
|
|
prevalidator_key=company_data.prevalidator_key,
|
|
seventh_amendment=company_data.seventh_amendment,
|
|
)
|
|
|
|
self.db.add(db_company)
|
|
self.db.commit()
|
|
self.db.refresh(db_company)
|
|
|
|
return CompanyResponseDTO.model_validate(db_company)
|
|
|
|
except IntegrityError as e:
|
|
self.db.rollback()
|
|
logger.error(f"IntegrityError creating company: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Integrity error: A company already exists in the system",
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Error creating company: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Error creating company")
|
|
|
|
def get_company(self) -> Optional[CompanyResponseDTO]:
|
|
"""
|
|
Obtiene la empresa (solo puede haber una)
|
|
|
|
Returns:
|
|
CompanyResponseDTO o None si no existe
|
|
"""
|
|
company = self.db.query(Company).filter(Company.id).first()
|
|
if not company:
|
|
return None
|
|
return CompanyResponseDTO.model_validate(company)
|
|
|
|
def get_company_by_id(self, company_id: str) -> Optional[CompanyResponseDTO]:
|
|
"""
|
|
Obtiene una empresa por ID
|
|
|
|
Args:
|
|
company_id: ID de la empresa
|
|
|
|
Returns:
|
|
CompanyResponseDTO o None si no existe
|
|
"""
|
|
company = self.db.query(Company).filter(Company.id == company_id).first()
|
|
if not company:
|
|
return None
|
|
return CompanyResponseDTO.model_validate(company)
|
|
|
|
def update_company(
|
|
self, company_id: str, company_data: CompanyUpdateDTO
|
|
) -> Optional[CompanyResponseDTO]:
|
|
"""
|
|
Actualiza una empresa
|
|
|
|
Args:
|
|
company_id: ID de la empresa a actualizar
|
|
company_data: Datos a actualizar
|
|
|
|
Returns:
|
|
CompanyResponseDTO actualizada o None si no existe
|
|
"""
|
|
company = self.db.query(Company).filter(Company.id == company_id).first()
|
|
if not company:
|
|
return None
|
|
|
|
# Actualizar solo campos proporcionados
|
|
update_data = company_data.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(company, field, value)
|
|
|
|
try:
|
|
self.db.commit()
|
|
self.db.refresh(company)
|
|
return CompanyResponseDTO.model_validate(company)
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Error updating company {company_id}: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Error updating company")
|
|
|
|
def delete_company(self, company_id: str) -> bool:
|
|
"""
|
|
Elimina una empresa
|
|
|
|
Args:
|
|
company_id: ID de la empresa a eliminar
|
|
|
|
Returns:
|
|
True si se eliminó, False si no existe
|
|
"""
|
|
company = self.db.query(Company).filter(Company.id == company_id).first()
|
|
if not company:
|
|
return False
|
|
|
|
try:
|
|
self.db.delete(company)
|
|
self.db.commit()
|
|
return True
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
logger.error(f"Error deleting company {company_id}: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Error deleting company")
|
|
|
|
def exists_company(self) -> bool:
|
|
"""
|
|
Verifica si existe una empresa registrada
|
|
|
|
Returns:
|
|
True si existe una empresa, False en caso contrario
|
|
"""
|
|
return self.db.query(Company).filter(Company.id).first() is not None
|
|
|
|
def get_companies_by_tenant(self, tenant_id: int) -> List[CompanyResponseDTO]:
|
|
"""
|
|
Obtiene todas las compañías que pertenecen a un tenant específico
|
|
|
|
Args:
|
|
tenant_id: ID del tenant
|
|
|
|
Returns:
|
|
Lista de CompanyResponseDTO
|
|
"""
|
|
companies = (
|
|
self.db.query(Company)
|
|
.filter(Company.tenant_id == tenant_id)
|
|
.order_by(Company.name)
|
|
.all()
|
|
)
|
|
|
|
return [CompanyResponseDTO.model_validate(company) for company in companies]
|