work-statistics/backend/app/services/supplier_service.py

"""Supplier service for business logic operations."""
from sqlalchemy.exc import IntegrityError
from app import db
from app.models.supplier import Supplier
from app.utils.validators import is_valid_name


class SupplierService:
    """Service class for Supplier CRUD operations."""
    
    @staticmethod
    def create(name):
        """Create a new supplier.
        
        Args:
            name: Supplier's name
            
        Returns:
            Tuple of (supplier_dict, error_message)
            On success: (supplier_dict, None)
            On failure: (None, error_message)
        """
        if not is_valid_name(name):
            return None, "供应商名称不能为空"
        
        # Check for duplicate name
        existing = Supplier.query.filter(Supplier.name == name.strip()).first()
        if existing:
            return None, "供应商名称已存在"
        
        try:
            supplier = Supplier(name=name.strip())
            db.session.add(supplier)
            db.session.commit()
            return supplier.to_dict(), None
        except IntegrityError:
            db.session.rollback()
            return None, "供应商名称已存在"
    
    @staticmethod
    def get_all():
        """Get all suppliers.
        
        Returns:
            List of supplier dictionaries
        """
        suppliers = Supplier.query.order_by(Supplier.id).all()
        return [s.to_dict() for s in suppliers]
    
    @staticmethod
    def get_by_id(supplier_id):
        """Get a supplier by ID.
        
        Args:
            supplier_id: Supplier's ID
            
        Returns:
            Tuple of (supplier_dict, error_message)
            On success: (supplier_dict, None)
            On failure: (None, error_message)
        """
        supplier = db.session.get(Supplier, supplier_id)
        if not supplier:
            return None, f"未找到ID为 {supplier_id} 的供应商"
        
        return supplier.to_dict(), None
    
    @staticmethod
    def update(supplier_id, name=None):
        """Update a supplier's name.
        
        Args:
            supplier_id: Supplier's ID
            name: New name (optional)
            
        Returns:
            Tuple of (supplier_dict, error_message)
            On success: (supplier_dict, None)
            On failure: (None, error_message)
        """
        supplier = db.session.get(Supplier, supplier_id)
        if not supplier:
            return None, f"未找到ID为 {supplier_id} 的供应商"
        
        if name is not None:
            if not is_valid_name(name):
                return None, "供应商名称不能为空"
            
            # Check for duplicate name (excluding current supplier)
            existing = Supplier.query.filter(
                Supplier.name == name.strip(),
                Supplier.id != supplier_id
            ).first()
            if existing:
                return None, "供应商名称已存在"
            
            try:
                supplier.name = name.strip()
                db.session.commit()
            except IntegrityError:
                db.session.rollback()
                return None, "供应商名称已存在"
        
        return supplier.to_dict(), None
    
    @staticmethod
    def delete(supplier_id):
        """Delete a supplier.
        
        Args:
            supplier_id: Supplier's ID
            
        Returns:
            Tuple of (success, error_message)
            On success: (True, None)
            On failure: (False, error_message)
        """
        supplier = db.session.get(Supplier, supplier_id)
        if not supplier:
            return False, f"未找到ID为 {supplier_id} 的供应商"
        
        # Check if supplier has associated items
        if hasattr(supplier, 'items') and supplier.items.count() > 0:
            return False, "无法删除已有关联物品的供应商"
        
        db.session.delete(supplier)
        db.session.commit()
        
        return True, None