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

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

from fastapi import APIRouter, Depends, UploadFile
from fastapi.responses import StreamingResponse, JSONResponse
from app.common.response import SuccessResponse, StreamResponse
from app.core.dependencies import AuthPermission
from app.core.router_class import OperationLogRoute
from app.api.v1.module_system.auth.schema import AuthSchema
from app.api.v1.module_system.user.schema import UserOutSchema
from app.common.request import PaginationService
from app.core.base_params import PaginationQueryParam
from app.utils.common_util import bytes2file_response

from {{ package_name }}.service.{{ table_name }}_service import {{ table_name|snake_to_pascal_case }}Service
from {{ package_name }}.schema import {{ table_name|snake_to_pascal_case }}CreateSchema, {{ table_name|snake_to_pascal_case }}UpdateSchema
from {{ package_name }}.param import {{ table_name|snake_to_pascal_case }}QueryParam

{{ table_name }}Controller = APIRouter(route_class=OperationLogRoute, prefix='/{{ module_name }}/{{ business_name }}', tags=["{{ function_name }}模块"])


@{{ table_name }}Controller.get('/list', summary="查询{{ function_name }}列表", description="查询{{ function_name }}列表")
async def get_{{ table_name }}_list(
        page: PaginationQueryParam = Depends(),
        search: {{ table_name|snake_to_pascal_case }}QueryParam = Depends(),
        auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:list"]))
):
    {{ table_name }}_result_list = await {{ table_name|snake_to_pascal_case }}Service.get_{{ table_name }}_list_service(auth, search, page.order_by)
    {{ table_name }}_result = await PaginationService.paginate(
        data_list={{ table_name }}_result_list, 
        page_no=page.page_no, 
        page_size=page.page_size
    )
    return SuccessResponse(data={{ table_name }}_result)


@{{ table_name }}Controller.get('/{id}', summary="获取{{ function_name }}详细信息", description="获取{{ function_name }}详细信息")
async def get_{{ table_name }}_by_id(
        id: int,
        auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:query"]))
):
    {{ table_name }} = await {{ table_name|snake_to_pascal_case }}Service.get_{{ table_name }}_by_id_service(auth, id)
    return SuccessResponse(data={{ table_name }})


@{{ table_name }}Controller.post('', summary="新增{{ function_name }}", description="新增{{ function_name }}")
async def add_{{ table_name }} (
    add_model: {{ table_name|snake_to_pascal_case }}CreateSchema,
    auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:add"])),
    current_user: UserOutSchema = Depends(lambda auth: auth.user)
):
    add_result = await {{ table_name|snake_to_pascal_case }}Service.add_{{ table_name }}_service(auth, add_model)
    return SuccessResponse(msg="新增成功")


@{{ table_name }}Controller.put('/{id}', summary="修改{{ function_name }}", description="修改{{ function_name }}")
async def update_{{ table_name }}(
    id: int,
    edit_model: {{ table_name|snake_to_pascal_case }}UpdateSchema,
    auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:edit"])),
    current_user: UserOutSchema = Depends(lambda auth: auth.user)
):
    update_result = await {{ table_name|snake_to_pascal_case }}Service.update_{{ table_name }}_service(auth, id, edit_model)
    return SuccessResponse(msg="修改成功")


@{{ table_name }}Controller.delete('/{ids}', summary="删除{{ function_name }}", description="删除{{ function_name }}")
async def del_{{ table_name }}(
    ids: str,
    auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:remove"]))
):
    id_list = [int(id) for id in ids.split(',')]
    del_result = await {{ table_name|snake_to_pascal_case }}Service.del_{{ table_name }}_service(auth, id_list)
    return SuccessResponse(msg="删除成功")


@{{ table_name }}Controller.post('/export', summary="导出{{ function_name }}", description="导出{{ function_name }}")
async def export_{{ table_name }}(
    search: {{ table_name|snake_to_pascal_case }}QueryParam = Depends(),
    auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:export"]))
) -> StreamingResponse:
    # 获取全量数据
    {{ table_name }}_list = await {{ table_name|snake_to_pascal_case }}Service.get_{{ table_name }}_list_service(auth=auth, search=search)
    export_result = await {{ table_name|snake_to_pascal_case }}Service.export_{{ table_name }}_list_service({{ table_name }}_list)
    return StreamResponse(
        data=bytes2file_response(export_result),
        media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
        headers = {
            'Content-Disposition': 'attachment; filename={{ table_name }}.xlsx'
        }
    )


@{{ table_name }}Controller.post('/import', summary="导入{{ function_name }}", description="导入{{ function_name }}")
async def import_{{ table_name }}(
    file: UploadFile,
    auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:import"]))
) -> JSONResponse:
    batch_import_result = await {{ table_name|snake_to_pascal_case }}Service.import_{{ table_name }}_service(file=file, auth=auth, update_support=True)
    return SuccessResponse(data=batch_import_result, msg="导入成功")


@{{ tableName }}Controller.post('/download/template', summary="获取{{ functionName }}导入模板", description="获取{{ functionName }}导入模板")
async def export_{{ tableName }}_template(
    auth: AuthSchema = Depends(AuthPermission(["{{ permissionPrefix }}:import"]))
) -> StreamingResponse:
    {{ tableName }}_import_template_result = await {{ tableName|snake_to_pascal_case }}Service.get_import_template_{{ tableName }}_service()
    return StreamResponse(
        data=bytes2file_response({{ tableName }}_import_template_result),
        media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
        headers = {
            'Content-Disposition': f'attachment; filename={{ tableName }}_import_template.xlsx'
        }
    )