FastapiAdmin/backend/templates/python/service.py.j2

# -*- coding:utf-8 -*-

from datetime import datetime
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Any, List, Dict, Optional

from app.core.exceptions import CustomException
from app.common.response import SuccessResponse
from app.api.v1.module_system.user.schema import UserOutSchema
from app.api.v1.module_system.auth.schema import AuthSchema
from {{ packageName }}.entity.vo.{{ tableName }}_vo import {{ tableName|snake_to_pascal_case }}PageModel, {{ tableName|snake_to_pascal_case }}Model
from {{ packageName }}.dao.{{ tableName }}_dao import {{ tableName|snake_to_pascal_case }}Dao


class {{ tableName|snake_to_pascal_case }}Service:
    """
    {{ functionName }}服务层
    """

    @classmethod
    async def get_{{ tableName }}_list_services(
        cls, auth: AuthSchema, query_object: {{ tableName|snake_to_pascal_case }}PageModel
    ):
        """
        获取{{ functionName }}列表信息service

        :param auth: 认证信息
        :param query_object: 查询参数对象
        :return: {{ functionName }}列表信息对象
        """
        # 确保db是AsyncSession类型
        if not isinstance(auth.db, AsyncSession):
            raise CustomException(msg='数据库会话类型不正确')
            
        {{ tableName }}_dao = {{ tableName|snake_to_pascal_case }}Dao(auth=auth)
        {{ tableName }}_list_result = await {{ tableName }}_dao.get_{{ tableName }}_list(auth.db, query_object, is_page=True)

        return {{ tableName }}_list_result

    @classmethod
    async def get_{{ tableName }}_by_id_services(cls, auth: AuthSchema, {{ tableName }}_id: int) -> {{ tableName|snake_to_pascal_case }}Model:
        """
        根据{{ tableName }}id获取{{ functionName }}信息service

        :param auth: 认证信息
        :param {{ tableName }}_id: {{ tableName }}id
        :return: {{ functionName }}信息对象
        """
        # 确保db是AsyncSession类型
        if not isinstance(auth.db, AsyncSession):
            raise CustomException(msg='数据库会话类型不正确')
            
        {{ tableName }}_dao = {{ tableName|snake_to_pascal_case }}Dao(auth=auth)
        {{ tableName }} = await {{ tableName }}_dao.get_{{ tableName }}_by_id(auth.db, {{ tableName }}_id)
        if {{ tableName }}:
            return {{ tableName }}
        else:
            raise CustomException(msg='{{ functionName }}不存在')

    @classmethod
    async def add_{{ tableName }}_services(
        cls, auth: AuthSchema, page_object: {{ tableName|snake_to_pascal_case }}Model
    ) -> SuccessResponse:
        """
        新增{{ functionName }}service

        :param auth: 认证信息
        :param page_object: 新增{{ functionName }}对象
        :return: 新增{{ functionName }}结果
        """
        # 确保db是AsyncSession类型
        if not isinstance(auth.db, AsyncSession):
            raise CustomException(msg='数据库会话类型不正确')
            
        {{ tableName }}_dao = {{ tableName|snake_to_pascal_case }}Dao(auth=auth)
        
        try:
            page_object.create_time = datetime.now()
            await {{ tableName }}_dao.create(data=page_object.model_dump())
            if isinstance(auth.db, AsyncSession):
                await auth.db.commit()
            return SuccessResponse(msg='新增成功')
        except Exception as e:
            if isinstance(auth.db, AsyncSession):
                try:
                    await auth.db.rollback()
                except:
                    pass  # 忽略回滚错误
            raise CustomException(msg=f'新增失败: {str(e)}')

    @classmethod
    async def update_{{ tableName }}_services(cls, auth: AuthSchema, page_object: {{ tableName|snake_to_pascal_case }}Model) -> SuccessResponse:
        """
        编辑{{ functionName }}service

        :param auth: 认证信息
        :param page_object: 编辑{{ functionName }}对象
        :return: 编辑{{ functionName }}结果
        """
        # 确保db是AsyncSession类型
        if not isinstance(auth.db, AsyncSession):
            raise CustomException(msg='数据库会话类型不正确')
            
        {{ tableName }}_dao = {{ tableName|snake_to_pascal_case }}Dao(auth=auth)
        
        # 检查必要字段是否存在
        if page_object.id is None:
            raise CustomException(msg='{{ functionName }}ID不能为空')
            
        edit_{{ tableName }} = page_object.model_dump(exclude_unset=True)
        {{ tableName }}_info = await cls.get_{{ tableName }}_by_id_services(auth, page_object.id)
        if {{ tableName }}_info:
            try:
                edit_{{ tableName }}['update_time'] = datetime.now()
                await {{ tableName }}_dao.update(id=page_object.id, data=edit_{{ tableName }})
                if isinstance(auth.db, AsyncSession):
                    await auth.db.commit()
                return SuccessResponse(msg='更新成功')
            except Exception as e:
                if isinstance(auth.db, AsyncSession):
                    try:
                        await auth.db.rollback()
                    except:
                        pass  # 忽略回滚错误
                raise CustomException(msg=f'更新失败: {str(e)}')
        else:
            raise CustomException(msg='{{ functionName }}不存在')

    @classmethod
    async def del_{{ tableName }}_services(cls, auth: AuthSchema, ids: List[str]) -> SuccessResponse:
        """
        删除{{ functionName }}service

        :param auth: 认证信息
        :param ids: {{ functionName }}id列表
        :return: 删除{{ functionName }}结果
        """
        # 确保db是AsyncSession类型
        if not isinstance(auth.db, AsyncSession):
            raise CustomException(msg='数据库会话类型不正确')
            
        {{ tableName }}_dao = {{ tableName|snake_to_pascal_case }}Dao(auth=auth)
        
        try:
            id_list = [int(id) for id in ids]
            await {{ tableName }}_dao.delete(ids=id_list)
            if isinstance(auth.db, AsyncSession):
                await auth.db.commit()
            return SuccessResponse(msg='删除成功')
        except Exception as e:
            if isinstance(auth.db, AsyncSession):
                try:
                    await auth.db.rollback()
                except:
                    pass  # 忽略回滚错误
            raise CustomException(msg=f'删除失败: {str(e)}')

    @classmethod
    async def export_{{ tableName }}_list_services(cls, auth: AuthSchema, query_object: {{ tableName|snake_to_pascal_case }}PageModel) -> bytes:
        """
        导出{{ functionName }}service

        :param auth: 认证信息
        :param query_object: 查询参数对象
        :return: 导出{{ functionName }}结果
        """
        # 确保db是AsyncSession类型
        if not isinstance(auth.db, AsyncSession):
            raise CustomException(msg='数据库会话类型不正确')
            
        {{ tableName }}_dao = {{ tableName|snake_to_pascal_case }}Dao(auth=auth)
        {{ tableName }}_list_result = await {{ tableName }}_dao.get_{{ tableName }}_list(auth.db, query_object, is_page=False)
        
        # 这里应该实现导出逻辑，例如生成Excel文件
        # 为简化起见，我们返回一个简单的文本文件
        export_data = "ID,名称\n"
        for item in {{ tableName }}_list_result.get("items", []):
            export_data += f"{item.id},{getattr(item, 'name', '')}\n"
        
        return export_data.encode('utf-8')


# -*- coding:utf-8 -*-

from typing import List
from sqlalchemy.ext.asyncio import AsyncSession
from utils.common_util import CamelCaseUtil, export_list2excel
from module_admin.entity.vo.sys_table_vo import SysTablePageModel
from module_admin.service.sys_table_service import SysTableService
from utils.page_util import PageResponseModel
from {{ packageName }}.dao.{{ tableName }}_dao import {{ tableName|snake_to_pascal_case }}Dao
from {{ packageName }}.entity.do.{{ tableName }}_do import {{ tableName|snake_to_pascal_case }}
from {{ packageName }}.entity.vo.{{ tableName }}_vo import {{ tableName|snake_to_pascal_case }}PageModel, {{ tableName|snake_to_pascal_case }}Model


class {{ tableName|snake_to_pascal_case }}Service:
    """
    {{ tableName|snake_to_pascal_case }}管理模块服务层
    """

    @classmethod
    async def get_{{ tableName }}_list(cls, query_db: AsyncSession, query_object: {{ tableName|snake_to_pascal_case }}PageModel, data_scope_sql: str) -> [list | PageResponseModel]:
        {{ tableName }}_list = await {{ tableName|snake_to_pascal_case }}Dao.get_{{ tableName }}_list(query_db, query_object, data_scope_sql, is_page=True)
        return {{ tableName }}_list

    @classmethod
    async def get_{{ tableName }}_by_id(cls, query_db: AsyncSession, {{ tableName }}_id: int) -> {{ tableName|snake_to_pascal_case }}Model:
        {{ tableName }} = await  {{ tableName|snake_to_pascal_case }}Dao.get_by_id(query_db, {{ tableName }}_id)
        {{ tableName }}_model = {{ tableName|snake_to_pascal_case }}Model(**CamelCaseUtil.transform_result({{ tableName }}))
        return {{ tableName }}_model


    @classmethod
    async def add_{{ tableName }}(cls, query_db: AsyncSession, query_object: {{ tableName|snake_to_pascal_case }}Model) -> {{ tableName|snake_to_pascal_case }}Model:
        {{ tableName }}_model = await {{ tableName|snake_to_pascal_case }}Dao.add_{{ tableName }}(query_db, query_object)
        return {{ tableName }}_model


    @classmethod
    async def update_{{ tableName }}(cls, query_db: AsyncSession, query_object: {{ tableName|snake_to_pascal_case }}Model) -> {{ tableName|snake_to_pascal_case }}Model:
        {{ tableName }} = await {{ tableName|snake_to_pascal_case }}Dao.edit_{{ tableName }}(query_db, query_object)
        {{ tableName }}_model = {{ tableName|snake_to_pascal_case }}Model(**CamelCaseUtil.transform_result({{ tableName }}))
        return {{ tableName }}_model


    @classmethod
    async def del_{{ tableName }}(cls, query_db: AsyncSession, {{ tableName }}_ids: List[str]):
        await {{ tableName|snake_to_pascal_case }}Dao.del_{{ tableName }}(query_db, {{ tableName }}_ids)


    @classmethod
    async def export_{{ tableName }}_list(cls, query_db: AsyncSession, query_object: {{ tableName|snake_to_pascal_case }}PageModel, data_scope_sql) -> bytes:
        {{ tableName }}_list = await {{ tableName|snake_to_pascal_case }}Dao.get_{{ tableName }}_list(query_db, query_object, data_scope_sql, is_page=False)
        filed_list = await SysTableService.get_sys_table_list(query_db, SysTablePageModel(tableName='{{ tableName }}'), is_page=False)
        filtered_filed = sorted(filter(lambda x: x["show"] == '1', filed_list), key=lambda x: x["sequence"])
        new_data = []
        for item in {{ tableName }}_list:
            mapping_dict = {}
            for fild in filtered_filed:
                if fild["prop"] in item:
                    mapping_dict[fild["label"]] = item[fild["prop"]]
            new_data.append(mapping_dict)
        binary_data = export_list2excel(new_data)
        return binary_data