""" Capa de servicio para lógica de negocio de empresa """ from sqlalchemy.orm import Session from sqlalchemy.exc import IntegrityError from fastapi import HTTPException from typing import List, Optional import logging from .models import Company from .dto import CompanyCreateDTO, CompanyUpdateDTO, CompanyResponseDTO 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.consecutive == True).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) logger.info(f"Company created: {db_company.id} - {db_company.name}") 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.consecutive == True).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) logger.info(f"Company updated: {company_id}") 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() logger.info(f"Company deleted: {company_id}") 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.consecutive == True).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]