mtsd/backend/app/crud/unit.py

"""商品单位CRUD"""
from typing import Optional, List
from sqlalchemy.orm import Session
from sqlalchemy import and_

from app.models.models import BasProductUnit, BasProductUnitHis
from app.crud.base import CRUDBase


class UnitCRUD(CRUDBase[BasProductUnit]):
    """商品单位CRUD操作"""

    def get_by_code(self, db: Session, code: str) -> Optional[BasProductUnit]:
        """根据编码获取单位"""
        return db.query(BasProductUnit).filter(
            and_(BasProductUnit.code == code, BasProductUnit.is_del == 0)
        ).first()

    def get_active_all(self, db: Session) -> List[BasProductUnit]:
        """获取所有启用的单位"""
        return db.query(BasProductUnit).filter(
            and_(BasProductUnit.is_del == 0, BasProductUnit.is_active == True)
        ).order_by(BasProductUnit.sort_order.asc(), BasProductUnit.id.asc()).all()

    def update_with_history(self, db: Session, id: str, obj_in: dict, updater: dict = None) -> Optional[BasProductUnit]:
        """更新单位并记录历史"""
        original = self.get(db, id)
        if not original:
            return None

        his_data = {
            "id": original.id,
            "name": original.name,
            "code": original.code,
            "is_active": original.is_active,
            "sort_order": original.sort_order,
            "create_at": original.create_at,
            "create_by": original.create_by,
            "create_name": original.create_name,
            "update_at": original.update_at,
            "update_by": original.update_by,
        }
        his = BasProductUnitHis(**his_data)
        db.add(his)

        return self.update(db, id, obj_in, updater)

    def soft_delete_with_history(self, db: Session, id: str) -> bool:
        """删除单位并记录历史"""
        original = self.get(db, id)
        if not original:
            return False

        his_data = {
            "id": original.id,
            "name": original.name,
            "code": original.code,
            "is_active": original.is_active,
            "sort_order": original.sort_order,
            "create_at": original.create_at,
            "create_by": original.create_by,
            "create_name": original.create_name,
            "update_at": original.update_at,
            "update_by": original.update_by,
        }
        his = BasProductUnitHis(**his_data)
        db.add(his)

        return self.delete(db, id)


unit_crud = UnitCRUD(BasProductUnit)