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

from sqlalchemy import delete, select, update, desc
from sqlalchemy.ext.asyncio import AsyncSession
from module_admin.entity.do.metasecurity_do import MetaSecurityCol, MetaSecurityRow
from module_admin.entity.vo.metasecurity_vo import MetaSecurityColModel, MetaSecurityRowModel
from utils.page_util import PageUtil


class MetaSecurityDao:
    """
    数据源安全管理模块数据库操作层
    """

    @classmethod
    async def get_meta_security_col_list(cls, db: AsyncSession, query_object: MetaSecurityColModel, is_page: bool = False):
     """
     获取 MetaSecurityCol 的列表信息，支持模糊查询和分页
     :param db: ORM对象
     :param query_object: 查询参数对象
     :param is_page: 是否开启分页
     :return: 列表信息
     """
     # 构建查询条件
     filters = []
     if query_object.dbCName:
         filters.append(MetaSecurityCol.dbCName.like(f"%{query_object.dbCName}%"))
     if query_object.obj_value:
         filters.append(MetaSecurityCol.obj_value.like(f"%{query_object.obj_value}%"))
     if query_object.dbTName:
         filters.append(MetaSecurityCol.dbTName.like(f"%{query_object.dbTName}%"))
     if query_object.dbSName:
         filters.append(MetaSecurityCol.dbSName.like(f"%{query_object.dbSName}%"))
     if query_object.dbRID:
         filters.append(MetaSecurityCol.dbRID==query_object.dbRID)         
     # 构建查询语句
     query = (
         select(MetaSecurityCol)
         .where(*filters)
         .order_by(desc(MetaSecurityCol.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_meta_security_row_list(cls, db: AsyncSession, query_object: MetaSecurityRowModel, is_page: bool = False):
      """
      获取 MetaSecurityRow 的列表信息，支持模糊查询和分页
      :param db: ORM对象
      :param query_object: 查询参数对象
      :param is_page: 是否开启分页
      :return: 列表信息
      """
      # 构建查询条件
      filters = []
      if query_object.dbCName:
          filters.append(MetaSecurityRow.dbCName.like(f"%{query_object.dbCName}%"))
      if query_object.obj_value:
          filters.append(MetaSecurityRow.obj_value.like(f"%{query_object.obj_value}%"))
      if query_object.dbTName:
          filters.append(MetaSecurityRow.dbTName.like(f"%{query_object.dbTName}%"))
      if query_object.dbSName:
         filters.append(MetaSecurityRow.dbSName.like(f"%{query_object.dbSName}%"))
      if query_object.dbRID:
         filters.append(MetaSecurityRow.dbRID==query_object.dbRID)              
      # 构建查询语句
      query = (
          select(MetaSecurityRow)
          .where(*filters)
          .order_by(desc(MetaSecurityRow.create_time))  # 按创建时间降序排序
      )
  
      # 分页处理
      row_list = await PageUtil.paginate(
          db, query, query_object.page_num, query_object.page_size, is_page
      )
      return row_list
  

    @classmethod
    async def get_meta_security_col_by_id(cls, db: AsyncSession, colId: str):
        col = (
            await db.execute(
                select(MetaSecurityCol)
                .where(MetaSecurityCol.colId == colId)
            )
        ).scalars().first()
        return col

    @classmethod
    async def get_meta_security_row_by_id(cls, db: AsyncSession, rowId: str):
        row = (
            await db.execute(
                select(MetaSecurityRow)
                .where(MetaSecurityRow.rowId == rowId)
            )
        ).scalars().first()
        return row

    @classmethod
    async def add_meta_security_col(cls, db: AsyncSession,colModel:MetaSecurityColModel):
        col = MetaSecurityCol(
           **colModel.model_dump()
        )
        db.add(col)
        await db.flush()
        return col
    @classmethod
    async def add_meta_security_row(cls, db: AsyncSession,rowModel:MetaSecurityRowModel):
        row = MetaSecurityRow(**rowModel.model_dump())
        db.add(row)
        await db.flush()
        return row

    @classmethod
    async def delete_meta_security_col(cls, db: AsyncSession, colId: str):
        await db.execute(delete(MetaSecurityCol).where(MetaSecurityCol.colId == colId))

    @classmethod
    async def delete_meta_security_row(cls, db: AsyncSession, rowId: str):
        await db.execute(delete(MetaSecurityRow).where(MetaSecurityRow.rowId == rowId))

    @classmethod
    async def update_meta_security_col(cls, db: AsyncSession, update_data: MetaSecurityColModel):
    
        await db.execute(update(MetaSecurityCol), [update_data])
        await db.flush()

    @classmethod
    async def update_meta_security_row(cls, db: AsyncSession,  update_data: MetaSecurityRowModel):
       await db.execute(update(MetaSecurityRow), [update_data])
       await db.flush()