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

from sqlalchemy import delete, select, update, desc
from sqlalchemy.ext.asyncio import AsyncSession
from module_admin.entity.do.datastd_do import DataStdCode
from module_admin.entity.vo.datastd_vo import DataStdCodeModel
from utils.page_util import PageUtil


class DataStdDao:
    """
    数据标准模块
    """

    @classmethod
    async def get_std_code_list(cls, db: AsyncSession, query_object: DataStdCodeModel, is_page: bool = False):
     """
     获取 DataStdCode 的列表信息，支持模糊查询和分页
     :param db: ORM对象
     :param query_object: 查询参数对象
     :param is_page: 是否开启分页
     :return: 列表信息
     """
     # 构建查询条件
     filters = []
     if query_object.code_name:
         filters.append(DataStdCode.code_name.like(f"%{query_object.code_name}%"))
     if query_object.code_status:
         filters.append(DataStdCode.code_status==query_object.code_status)
     if query_object.sys_id:
         filters.append(DataStdCode.sys_id==query_object.sys_id)      
     if query_object.code_type:
         filters.append(DataStdCode.code_type==query_object.code_type) 
         
     if query_object.class_id:
         filters.append(DataStdCode.class_id==query_object.class_id) 
         if query_object.class_id=='codeItem' and not query_object.parent_id:
                filters.append(1==2) 
     if query_object.parent_id:
         filters.append(DataStdCode.parent_id==query_object.parent_id)             
     # 构建查询语句
     query = (
         select(DataStdCode)
         .where(*filters)
         .order_by(desc(DataStdCode.create_time))  # 按创建时间降序排序
     )

     # 分页处理
     col_list = await PageUtil.paginate(
         db, query, query_object.page_num, query_object.page_size, is_page
     )
     return col_list


 
    @classmethod
    async def get_data_code_list_by_info(cls, db: AsyncSession, query_object: DataStdCodeModel):
        List = (
            await db.execute(
                select(DataStdCode)
                .where(
                        DataStdCode.code_name == query_object.code_name if query_object.code_name else True,
                        DataStdCode.code_status == query_object.code_status if query_object.code_status else True,
                        DataStdCode.sys_id == query_object.sys_id if query_object.sys_id else True,
                        DataStdCode.code_type == query_object.code_type if query_object.code_type else True,
                        DataStdCode.parent_id == query_object.parent_id if query_object.parent_id else True,
                        DataStdCode.class_id == query_object.class_id if query_object.class_id else True
                )
            )
        ).scalars().all()
        return List
    @classmethod
    async def get_data_code_by_info(cls, db: AsyncSession, query_object: DataStdCodeModel):
        List = (
            await db.execute(
                select(DataStdCode)
                .where(
                        DataStdCode.code_name == query_object.code_name if query_object.code_name else True,
                        DataStdCode.code_num == query_object.code_num if query_object.code_num else True,
                        DataStdCode.code_status == query_object.code_status if query_object.code_status else True,
                        DataStdCode.sys_id == query_object.sys_id if query_object.sys_id else True,
                        DataStdCode.code_type == query_object.code_type if query_object.code_type else True,
                        DataStdCode.parent_id == query_object.parent_id if query_object.parent_id else True,
                        DataStdCode.class_id == query_object.class_id if query_object.class_id else True
                )
            )
        ).scalars().first()
        return List
    @classmethod
    async def get_std_code_by_id(cls, db: AsyncSession, Id: str):
        col = (
            await db.execute(
                select(DataStdCode)
                .where(DataStdCode.id == Id)
            )
        ).scalars().first()
        return col


    @classmethod
    async def add_std_code(cls, db: AsyncSession,model:DataStdCodeModel):
        col = DataStdCode(
           **model.model_dump()
        )
        db.add(col)
        await db.flush()
        return col

    @classmethod
    async def delete_std_code(cls, db: AsyncSession, Id: str):
        await db.execute(delete(DataStdCode).where(DataStdCode.id == Id))

 
    @classmethod
    async def update_std_code(cls, db: AsyncSession, update_data: DataStdCodeModel):
    
        await db.execute(update(DataStdCode), [update_data])
        await db.flush()

    @classmethod
    async def check_code_num_exists(cls, query_db: AsyncSession, code_num: str) -> bool:
        """
        检查标准代码或代码项的code_num是否已经存在
        """
        # 查询标准代码表中是否存在相同的code_num
        result = await query_db.execute(select(DataStdCodeModel).filter(DataStdCodeModel.code_num == code_num))
        existing_code = result.scalar_one_or_none()

        # 如果不存在，则检查代码项表中是否存在相同的code_num
        if not existing_code:
            result = await query_db.execute(select(DataStdCodeModel).filter(DataStdCodeModel.code_num == code_num))
            existing_code_item = result.scalar_one_or_none()
            return existing_code_item is not None

        return True