shanxia/bx/bx-backend/app/services/expense_item.py

from sqlalchemy.orm import Session

from app.models.expense_item import ExpenseItem
from app.services.base import CrudService


class ExpenseItemService:

    @staticmethod
    def get_by_id(db: Session, id: int) -> ExpenseItem | None:
        return CrudService.get_by_id(ExpenseItem, db, id)

    @staticmethod
    def get_paged(db: Session, page: int, page_size: int, filters: list | None = None):
        return CrudService.get_paged(ExpenseItem, db, page, page_size, filters)

    @staticmethod
    def create(db: Session, obj_in: dict) -> ExpenseItem:
        return CrudService.create(ExpenseItem, db, obj_in)

    @staticmethod
    def update(db: Session, id: int, obj_in: dict) -> ExpenseItem | None:
        return CrudService.update(ExpenseItem, db, id, obj_in)

    @staticmethod
    def soft_delete(db: Session, id: int) -> bool:
        return CrudService.soft_delete(ExpenseItem, db, id)

    @staticmethod
    def get_item_tree(db: Session) -> list[ExpenseItem]:
        items = (
            db.query(ExpenseItem)
            .filter(ExpenseItem.del_flag == 0)
            .order_by(ExpenseItem.sort_order)
            .all()
        )
        return ExpenseItemService._build_tree(items)

    @staticmethod
    def _build_tree(items: list[ExpenseItem]) -> list[ExpenseItem]:
        item_map = {item.id: item for item in items}
        roots: list[ExpenseItem] = []
        for item in items:
            if not item.parent_id or item.parent_id == 0:
                roots.append(item)
            else:
                parent = item_map.get(item.parent_id)
                if parent:
                    if not hasattr(parent, "children") or parent.children is None:
                        parent.children = []
                    parent.children.append(item)
        for item in items:
            if hasattr(item, "children") and item.children:
                item.child_count = len(item.children)
            else:
                item.child_count = 0
        return roots