payment-platform/backend/app/plugin/module_payment/department/crud.py

from app.api.v1.module_system.auth.schema import AuthSchema
from app.core.base_crud import CRUDBase
from app.core.exceptions import CustomException

from .model import DepartmentModel
from .schema import DepartmentCreateSchema


class DepartmentCRUD(CRUDBase[DepartmentModel, DepartmentCreateSchema, DepartmentCreateSchema]):
    """部门 CRUD 操作"""

    def __init__(self, auth: AuthSchema) -> None:
        self.auth = auth
        super().__init__(model=DepartmentModel, auth=auth)

    async def get_by_department_id(
        self, department_id: str
    ) -> DepartmentModel | None:
        """根据部门ID查询部门(业务主键)"""
        return await self.get(department_id=department_id)

    async def get_by_parent_id(
        self, parent_id: str | None
    ) -> list[DepartmentModel]:
        """根据父部门ID查询子部门列表"""
        if parent_id is None:
            return await self.get_all(parent_id=None)
        return await self.get_all(parent_id=parent_id)

    async def get_root_departments(self) -> list[DepartmentModel]:
        """获取根部门列表(parent_id为空的部门)"""
        return await self.get_all(parent_id=None)

    async def get_children(self, department_id: str) -> list[DepartmentModel]:
        """获取子部门列表"""
        return await self.get_all(parent_id=department_id)

    async def has_children(self, department_id: str) -> bool:
        """检查部门是否有子部门"""
        children = await self.get_children(department_id)
        return len(children) > 0

    async def update_by_department_id(
        self, department_id: str, data: dict
    ) -> DepartmentModel | None:
        """根据部门ID更新部门"""
        obj = await self.get(department_id=department_id, preload=[])
        if not obj:
            raise CustomException(msg="更新失败！对象不存在")

        if self.auth.user and hasattr(obj, "updated_id"):
            setattr(obj, "updated_id", self.auth.user.id)

        for key, value in data.items():
            if hasattr(obj, key):
                setattr(obj, key, value)

        await self.auth.db.flush()
        await self.auth.db.refresh(obj)

        verify_obj = await self.get(department_id=department_id, preload=[])
        if not verify_obj:
            raise CustomException(msg="更新失败！对象不存在或无权限访问")

        return obj