dash-fastapi-admin/dash-fastapi-backend/module_admin/dao/dscatalog_dao.py

from sqlalchemy import or_, func
from sqlalchemy.orm import Session
from sqlalchemy.sql.expression import text
from module_admin.entity.do.dscatalog_do import SysDscatalog
from module_admin.entity.vo.dscatalog_vo import DscatalogModel, DscatalogResponse, CrudDscatalogResponse
from utils.time_format_util import list_format_datetime


class DscatalogDao:
    """
    数据标准管理模块数据库操作层
    """

    @classmethod
    def get_dscatalog_by_id(cls, db: Session, dscatalog_id: int):
        """
        根据数据标准id获取在用数据标准信息
        :param db: orm对象
        :param dscatalog_id: 数据标准id
        :return: 在用数据标准信息对象
        """
        dscatalog_info = db.query(SysDscatalog) \
            .filter(SysDscatalog.dscatalog_id == dscatalog_id,
                    SysDscatalog.status == 0,
                    SysDscatalog.del_flag == 0) \
            .first()

        return dscatalog_info

    @classmethod
    def get_dscatalog_by_id_for_list(cls, db: Session, dscatalog_id: int):
        """
        用于获取数据标准列表的工具方法
        :param db: orm对象
        :param dscatalog_id: 数据标准id
        :return: 数据标准id对应的信息对象
        """
        dscatalog_info = db.query(SysDscatalog) \
            .filter(SysDscatalog.dscatalog_id == dscatalog_id,
                    SysDscatalog.del_flag == 0) \
            .first()

        return dscatalog_info

    @classmethod
    def get_dscatalog_detail_by_id(cls, db: Session, dscatalog_id: int):
        """
        根据数据标准id获取数据标准详细信息
        :param db: orm对象
        :param dscatalog_id: 数据标准id
        :return: 数据标准信息对象
        """
        dscatalog_info = db.query(SysDscatalog) \
            .filter(SysDscatalog.dscatalog_id == dscatalog_id,
                    SysDscatalog.del_flag == 0) \
            .first()

        return dscatalog_info

    @classmethod
    def get_dscatalog_detail_by_info(cls, db: Session, dscatalog: DscatalogModel):
        """
        根据数据标准参数获取数据标准信息
        :param db: orm对象
        :param dscatalog: 数据标准参数对象
        :return: 数据标准信息对象
        """
        dscatalog_info = db.query(SysDscatalog) \
            .filter(SysDscatalog.dscatalog_parent_id == dscatalog.dscatalog_parent_id if dscatalog.dscatalog_parent_id else True,
                    SysDscatalog.dscatalog_name == dscatalog.dscatalog_name if dscatalog.dscatalog_name else True) \
            .first()

        return dscatalog_info

    @classmethod
    def get_dscatalog_info_for_edit_option(cls, db: Session, dscatalog_info: DscatalogModel, data_scope_sql: str):
        """
        获取数据标准编辑对应的在用数据标准列表信息
        :param db: orm对象
        :param dscatalog_info: 数据标准对象
        :param data_scope_sql: 数据权限对应的查询sql语句
        :return: 数据标准列表信息
        """
        dscatalog_result = db.query(SysDscatalog) \
            .filter(SysDscatalog.dscatalog_id != dscatalog_info.dscatalog_id,
                    SysDscatalog.dscatalog_parent_id != dscatalog_info.dscatalog_id,
                    SysDscatalog.del_flag == 0, SysDscatalog.status == 0,
                    eval(data_scope_sql)) \
            .order_by(SysDscatalog.dscatalog_order_num) \
            .distinct().all()

        return list_format_datetime(dscatalog_result)

    @classmethod
    def get_children_dscatalog(cls, db: Session, dscatalog_id: int):
        """
        根据数据标准id查询当前数据标准的子数据标准列表信息
        :param db: orm对象
        :param dscatalog_id: 数据标准id
        :return: 子数据标准信息列表
        """
        dscatalog_result = db.query(SysDscatalog) \
            .filter(SysDscatalog.dscatalog_parent_id == dscatalog_id,
                    SysDscatalog.del_flag == 0) \
            .all()

        return list_format_datetime(dscatalog_result)

    @classmethod
    def get_dscatalog_all_ancestors(cls, db: Session):
        """
        获取所有数据标准的ancestors信息
        :param db: orm对象
        :return: ancestors信息列表
        """
        dscatalog_ancestors = db.query(SysDscatalog.dscatalog_ancestors)\
            .filter(SysDscatalog.del_flag == 0)\
            .all()

        return dscatalog_ancestors

    @classmethod
    def get_dscatalog_list_for_tree(cls, db: Session, dscatalog_info: DscatalogModel, data_scope_sql: str):
        """
        获取所有在用数据标准列表信息
        :param db: orm对象
        :param dscatalog_info: 数据标准对象
        :param data_scope_sql: 数据权限对应的查询sql语句
        :return: 在用数据标准列表信息
        """
        dscatalog_result = db.query(SysDscatalog) \
            .filter(SysDscatalog.status == 0,
                    SysDscatalog.del_flag == 0,
                    #SysDscatalog.dscatalog_area != '语句配置',
                    SysDscatalog.dscatalog_name.like(f'%{dscatalog_info.dscatalog_name}%') if dscatalog_info.dscatalog_name else True,
                    eval(data_scope_sql)) \
            .order_by(SysDscatalog.dscatalog_order_num) \
            .distinct().all()

        return list_format_datetime(dscatalog_result)

    @classmethod
    def get_dscatalog_list(cls, db: Session, page_object: DscatalogModel, data_scope_sql: str):
        """
        根据查询参数获取数据标准列表信息
        :param db: orm对象
        :param page_object: 不分页查询参数对象
        :param data_scope_sql: 数据权限对应的查询sql语句
        :return: 数据标准列表信息对象
        """
        dscatalog_result = db.query(SysDscatalog) \
            .filter(SysDscatalog.del_flag == 0,
                    SysDscatalog.status == page_object.status if page_object.status else True,
                    SysDscatalog.dscatalog_name.like(f'%{page_object.dscatalog_name}%') if page_object.dscatalog_name else True,
                    eval(data_scope_sql)) \
            .order_by(SysDscatalog.dscatalog_order_num) \
            .distinct().all()

        result = dict(
            rows=list_format_datetime(dscatalog_result),
        )

        return DscatalogResponse(**result)

    @classmethod
    def add_dscatalog_dao(cls, db: Session, dscatalog: DscatalogModel):
        """
        新增数据标准数据库操作
        :param db: orm对象
        :param dscatalog: 数据标准对象
        :return: 新增校验结果
        """
        db_dscatalog = SysDscatalog(**dscatalog.dict())
        db.add(db_dscatalog)
        db.commit()
        db.flush()


        # 构建原生SQL更新语句
        update_sql = """
        update sys_dscatalog a
        join (
            select 
			row_number() over(PARTITION BY a.dscatalog_parent_id order by a.dscatalog_id ) as s1, 
			LENGTH(dscatalog_ancestors) - LENGTH(REPLACE(dscatalog_ancestors, ',', '')) + 1 as s2,
			a.* from sys_dscatalog a
        ) b on a.dscatalog_id = b.dscatalog_id
        set a.dscatalog_order_num = b.s1,
		a.dscatalog_area = concat(b.s2,'级')
        """
        # 执行原生SQL
        db.execute(text(update_sql))
        db.commit()
        db.flush()



        return db_dscatalog

    @classmethod
    def edit_dscatalog_dao(cls, db: Session, dscatalog: dict):
        """
        编辑数据标准数据库操作
        :param db: orm对象
        :param dscatalog: 需要更新的数据标准字典
        :return: 编辑校验结果
        """
        db.query(SysDscatalog) \
            .filter(SysDscatalog.dscatalog_id == dscatalog.get('dscatalog_id')) \
            .update(dscatalog)

    @classmethod
    def delete_dscatalog_dao(cls, db: Session, dscatalog: DscatalogModel):
        """
        删除数据标准数据库操作
        :param db: orm对象
        :param dscatalog: 数据标准对象
        :return:
        #.update({SysDscatalog.del_flag: '2', SysDscatalog.update_by: dscatalog.update_by, SysDscatalog.update_time: dscatalog.update_time})
        """
        db.query(SysDscatalog) \
            .filter(SysDscatalog.dscatalog_id == dscatalog.dscatalog_id) \
            .delete()