from datetime import datetime, time from sqlalchemy import delete, select, update from sqlalchemy.ext.asyncio import AsyncSession from module_admin.entity.do.meta_do import MetadataClas # ORM 类 class MetadataConfigDao: """ 标签信息数据库操作层 """ @classmethod async def get_clas_detail_by_id(cls, db: AsyncSession, clas_id: int): """ 根据标签序号获取标签详细信息 """ result = await db.execute(select(MetadataClas).where(MetadataClas.clas_onum == clas_id)) return result.scalars().first() @classmethod async def get_clas_detail_by_info(cls, db: AsyncSession, clas): """ 根据标签参数获取标签信息(根据 MetadataClasModel 实例) """ result = await db.execute( select(MetadataClas).where( MetadataClas.clas_name == clas.clas_name if clas.clas_name else True, MetadataClas.clas_eff_flag == clas.clas_eff_flag if clas.clas_eff_flag else True, ) ) return result.scalars().first() @classmethod async def get_metadata_clas_list(cls, db: AsyncSession, query_object, is_page: bool = False): """ 获取标签信息列表(支持分页) """ query = select(MetadataClas).where( MetadataClas.clas_pri_clas.like(f"%{query_object.clas_pri_clas}%") if query_object.clas_pri_clas else True, MetadataClas.clas_scd_clas.like(f"%{query_object.clas_scd_clas}%") if query_object.clas_scd_clas else True, MetadataClas.clas_thre_clas.like(f"%{query_object.clas_thre_clas}%") if query_object.clas_thre_clas else True, MetadataClas.clas_name.like(f"%{query_object.clas_name}%") if query_object.clas_name else True, MetadataClas.clas_eff_flag == query_object.clas_eff_flag if query_object.clas_eff_flag else True, MetadataClas.upd_time.between( datetime.combine(datetime.strptime(query_object.begin_time, '%Y-%m-%d'), time(0, 0, 0)), datetime.combine(datetime.strptime(query_object.end_time, '%Y-%m-%d'), time(23, 59, 59)), ) if query_object.begin_time and query_object.end_time else True, ).order_by(MetadataClas.clas_onum).distinct() from utils.page_util import PageUtil return await PageUtil.paginate(db, query, query_object.page_num, query_object.page_size, is_page) @classmethod async def add_metadata_clas_dao(cls, db: AsyncSession, clas): """ 新增标签信息 """ db_clas = MetadataClas(**clas.model_dump()) db.add(db_clas) await db.flush() return db_clas @classmethod async def edit_metadata_clas_dao(cls, db: AsyncSession, clas_id: int, update_data: dict): """ 修改标签信息 """ await db.execute( update(MetadataClas) .where(MetadataClas.clas_onum == clas_id) .values(**update_data) ) @classmethod async def delete_metadata_clas_dao(cls, db: AsyncSession, clas_onum_list: list[int]): """ 删除标签信息(支持批量) """ await db.execute(delete(MetadataClas).where(MetadataClas.clas_onum.in_(clas_onum_list)))