VberAdminPy/VapAdmin/BACKEND/domain/services/admin/sys_dict_service.py

from typing import Optional, List

from sqlalchemy import select, func, Sequence

from core.enums import StatusTypeEnum
from core.exceptions import ServiceWarning
from domain.dtos import (
    SysDictTypeDto,
    SysDictTypeUpdateDto,
    SysDictDataDto,
    SysDictDataUpdateDto,
    PageResultDto,
    PageDto,
)
from domain.models import SysDictTypeModel, SysDictDataModel
from domain.services.base_services import (
    CurdServiceBase,
)


class SysDictService(
    CurdServiceBase[
        SysDictTypeModel, SysDictTypeDto, SysDictTypeUpdateDto, SysDictTypeUpdateDto
    ]
):
    def __init__(self, db_name: Optional[str] = None):
        super().__init__(
            SysDictTypeModel,
            SysDictTypeDto,
            SysDictTypeUpdateDto,
            SysDictTypeUpdateDto,
            db_name=db_name,
        )
        from domain.services import CommonService

        self._common_service = CommonService()

    async def get_dict_data_page(
        self, page_dto: PageDto
    ) -> PageResultDto[SysDictDataDto]:
        """根据字典类型查询字典数据"""
        try:
            async with await self._get_async_db() as db:
                query = select(SysDictDataModel).where(
                    SysDictDataModel.status == StatusTypeEnum.NORMAL.key,
                )
                query = self._apply_search_base(
                    SysDictDataModel, query, page_dto.search
                )
                query = self._apply_filter_base(
                    SysDictDataModel,
                    query,
                    page_dto.filters,
                    page_dto.filter_conditions,
                )
                total = await db.scalar(select(func.count()).select_from(query))
                query = query.order_by(page_dto.order)
                query = query.offset(page_dto.offset).limit(page_dto.limit)
                result = await db.execute(query)
                models = result.scalars().all()
                return PageResultDto(
                    total=total,
                    rows=self._apply_map_get_data_dto_list(models),
                    page_size=page_dto.page,
                )
        except Exception as e:
            raise ServiceWarning(f"分页查询字典数据失败: {str(e)}")

    @staticmethod
    def _apply_map_get_data_dto_list(
        models: Sequence[SysDictDataModel],
    ) -> List[SysDictDataDto]:
        """将模型列表映射为DTO列表"""
        return [SysDictDataDto.from_model(model) for model in models]

    async def get_all_dict_data(self) -> List[SysDictDataDto]:
        """根据字典类型查询字典数据"""
        result = await self._common_service.get_all_dict_data()
        return result

    async def create(self, obj_in: SysDictTypeUpdateDto) -> SysDictTypeModel:
        """空实现 - 创建操作禁用"""
        return SysDictTypeModel()

    async def delete(self, id: int) -> None:
        """空实现 - 删除操作禁用"""
        pass

    async def update_dict_data(self, obj_in: SysDictDataUpdateDto):
        """更新字典数据"""
        db_obj = self.get(obj_in.id)
        if not db_obj:
            raise ServiceWarning(f"字典数据不存在: {obj_in.id}")
        try:
            async with await self._get_async_db() as db:
                db_obj.dict_sort = obj_in.dict_sort
                db_obj.dict_label = obj_in.dict_label
                db_obj.css_class = obj_in.css_class
                db_obj.list_class = obj_in.list_class
                db_obj.status = obj_in.status
                db_obj.remark = obj_in.remark
                await db.commit()
                await db.refresh(db_obj)
                self._common_service.get_all_dict_data.cache_clear()
                return SysDictDataDto.from_model(db_obj)
        except Exception as e:
            raise ServiceWarning(f"更新字典数据失败: {str(e)}")