fastapi-best-architecture/backend/plugin/code_generator/templates/python/service.jinja

from typing import Any, Sequence

from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.{{ app_name }}.crud.crud_{{ table_name }} import {{ table_name }}_dao
from backend.app.{{ app_name }}.model import {{ class_name }}
from backend.app.{{ app_name }}.schema.{{ table_name }} import Create{{ schema_name }}Param, Delete{{ schema_name }}Param, Update{{ schema_name }}Param
from backend.common.exception import errors
from backend.common.pagination import paging_data


class {{ class_name }}Service:
    @staticmethod
    async def get(*, db: AsyncSession, pk: int) -> {{ class_name }}:
        """
        获取{{ doc_comment }}

        :param db: 数据库会话
        :param pk: {{ doc_comment }} ID
        :return:
        """
        {{ table_name }} = await {{ table_name }}_dao.get(db, pk)
        if not {{ table_name }}:
            raise errors.NotFoundError(msg='{{ doc_comment }}不存在')
        return {{ table_name }}

    @staticmethod
    async def get_list(db: AsyncSession) -> dict[str, Any]:
        """
        获取{{ doc_comment }}列表

        :param db: 数据库会话
        :return:
        """
        {{ table_name }}_select = await {{ table_name }}_dao.get_select()
        return await paging_data(db, {{ table_name }}_select)

    @staticmethod
    async def get_all(*, db: AsyncSession) -> Sequence[{{ class_name }}]:
        """
        获取所有{{ doc_comment }}

        :param db: 数据库会话
        :return:
        """
        {{ table_name }}s = await {{ table_name }}_dao.get_all(db)
        return {{ table_name }}s

    @staticmethod
    async def create(*, db: AsyncSession, obj: Create{{ schema_name }}Param) -> None:
        """
        创建{{ doc_comment }}

        :param db: 数据库会话
        :param obj: 创建{{ doc_comment }}参数
        :return:
        """
        await {{ table_name }}_dao.create(db, obj)

    @staticmethod
    async def update(*, db: AsyncSession, pk: int, obj: Update{{ schema_name }}Param) -> int:
        """
        更新{{ doc_comment }}

        :param db: 数据库会话
        :param pk: {{ doc_comment }} ID
        :param obj: 更新{{ doc_comment }}参数
        :return:
        """
        count = await {{ table_name }}_dao.update(db, pk, obj)
        return count

    @staticmethod
    async def delete(*, db: AsyncSession, obj: Delete{{ schema_name }}Param) -> int:
        """
        删除{{ doc_comment }}

        :param db: 数据库会话
        :param obj: {{ doc_comment }} ID 列表
        :return:
        """
        count = await {{ table_name }}_dao.delete(db, obj.pks)
        return count


{{ table_name }}_service: {{ class_name }}Service = {{ class_name }}Service()