vue-fastapi-admin/vue-fastapi-backend/module_admin/dao/approval_dao.py

from sqlalchemy import desc, delete, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from module_admin.entity.do.approval_do import FlowApproval, FlowConfig
from module_admin.entity.vo.approval_vo import ApprovalQueryObject
from module_admin.entity.vo.user_vo import CurrentUserModel
from sqlalchemy import select, text, cast, Integer, and_, or_, outerjoin, func, join
from utils.page_util import PageUtil


class ApprovalDao:
    """
    菜单管理模块数据库操作层
    """

    @classmethod
    async def add_flow_approval(cls, db: AsyncSession, flow_approval: FlowApproval):
        db.add(flow_approval)
        await db.flush()
        return flow_approval

    @classmethod
    async def get_flow_by_id(cls, db: AsyncSession, flowId: str):
        result = (
            (
                await db.execute(
                    select(FlowApproval).where(
                        FlowApproval.id == flowId
                    )
                )
            )
            .scalars()
            .first()
        )
        return result

    @classmethod
    async def edit_flow_approval(cls, db: AsyncSession, flow: dict):
        await db.execute(update(FlowApproval), [flow])

    @classmethod
    async def get_flow_list(cls, db: AsyncSession, query_param: ApprovalQueryObject, current_user: CurrentUserModel):
        roleList = current_user.roles
        query = (select(FlowApproval).where(
            (FlowApproval.applicant == query_param.applicant) if query_param.applicant else True,
            (FlowApproval.businessType == query_param.businessType) if query_param.businessType else True,
            or_(*[FlowApproval.nextStepRole.like(f'%{role}%') for role in roleList],
                FlowApproval.nextStepUser.like(f'%{current_user.user.user_name}%')),
        ).order_by(FlowApproval.applyTime)
                 .distinct())
        # 注意：这里不执行查询，而是将查询对象传递给 paginate 方法
        result = await PageUtil.paginate(db, query, query_param.page_num, query_param.page_size, True)
        return result

    @classmethod
    async def get_owner_flow_list(cls, db: AsyncSession, query_param: ApprovalQueryObject,
                                  current_user: CurrentUserModel):
        query = (
            select(FlowApproval)
            .where(
                (FlowApproval.applicant == query_param.applicant) if query_param.applicant else True,
                (FlowApproval.businessType == query_param.businessType) if query_param.businessType else True,
                FlowApproval.approvalFlow.like(f'%{current_user.user.user_name}%')
            )
            .order_by(FlowApproval.applyTime)
            .distinct()
        )
        # 注意：这里不执行查询，而是将查询对象传递给 paginate 方法
        result = await PageUtil.paginate(db, query, query_param.page_num, query_param.page_size, True)
        return result

    @classmethod
    async def get_my_flow_list(cls, db: AsyncSession, query_param: ApprovalQueryObject,
                               current_user: CurrentUserModel):
        query = (
            select(FlowApproval)
            .where(
                FlowApproval.applicant == current_user.user.user_name,
                (FlowApproval.businessType == query_param.businessType) if query_param.businessType else True,
            )
            .order_by(FlowApproval.applyTime)
            .distinct()
        )
        # 注意：这里不执行查询，而是将查询对象传递给 paginate 方法
        result = await PageUtil.paginate(db, query, query_param.page_num, query_param.page_size, True)
        return result

    @classmethod
    async def get_all_waitingOrPendingFlows(cls, db: AsyncSession):
        result = (
            (
                await db.execute(select(FlowApproval).where(
                    or_(FlowApproval.status == 'waiting',
                        FlowApproval.status == 'pending')
                ))
            ).scalars().all()
        )
        return result

    @classmethod
    async def get_flow_by_idAndUser(cls, db: AsyncSession, flow_id: str, username: str):
        result = (
            (
                await db.execute(
                    select(FlowApproval).where(
                        FlowApproval.id == flow_id,
                        FlowApproval.applicant == username
                    )
                )
            )
            .scalars()
            .first()
        )
        return result

    @classmethod
    async def get_conf_list(cls, db: AsyncSession, module: str):
        result = (
            (
                await db.execute(select(FlowConfig).where(
                    FlowConfig.module == module
                ))
            ).scalars().all()
        )
        return result

    @classmethod
    async def add_flow_config(cls, db: AsyncSession, conf: dict):
        db_conf = FlowConfig(**conf)
        db.add(db_conf)
        await db.flush()
        return conf

    @classmethod
    async def delete_flow_by_module(cls, module: str, db: AsyncSession):
        await db.execute(delete(FlowConfig).where(FlowConfig.module == module))

    @classmethod
    async def get_nextstep_conf(cls, module: str, db: AsyncSession, step: int):
        result = (
            (
                await db.execute(select(FlowConfig).where(
                    FlowConfig.module == module,
                    FlowConfig.step == step
                ))
            ).scalars().all()
        )
        return result

    @classmethod
    async def get_first_node_conf(cls, module: str, db: AsyncSession):
        result = (
            (
                await db.execute(select(FlowConfig).where(
                    FlowConfig.module == module,
                    FlowConfig.step == 1
                ))
            ).scalars().all()
        )
        return result