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

# -*- coding: utf-8 -*-
import urllib.parse
from typing import Annotated

from fastapi import APIRouter, Body, Depends, File, Path, Query, UploadFile, status
from fastapi.responses import JSONResponse, StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession

from app.common.response import ResponseSchema, StreamResponse, SuccessResponse
from app.core.base_schema import AuthSchema, BatchSetAvailable, PageResultSchema, PaginationQueryParam
from app.core.dependencies import AuthPermission, db_getter
from app.core.router_class import OperationLogRoute
from app.utils.common_util import bytes2file_response

from .schema import {{ class_name }}CreateSchema, {{ class_name }}OutSchema, {{ class_name }}QueryParam, {{ class_name }}UpdateSchema
from .service import {{ class_name }}Service

{{ class_name }}Router = APIRouter(route_class=OperationLogRoute, prefix="/{{ module_name }}", tags=["{{ function_name }}模块"])


@{{ class_name }}Router.get("/detail/{id}", summary="获取{{ function_name }}详情", response_model=ResponseSchema[{{ class_name }}OutSchema])
async def get_obj_detail_controller(
    auth: Annotated[AuthSchema, Depends(AuthPermission(["{{ permission_prefix }}:detail"]))],
    id: Annotated[int, Path(description="{{ function_name }}ID")],
    db: Annotated[AsyncSession, Depends(db_getter)],
) -> JSONResponse:
    service = {{ class_name }}Service(auth, db)
    result_dict = await service.detail(id=id)
    return SuccessResponse(data=result_dict, msg="获取{{ function_name }}详情成功")


@{{ class_name }}Router.get("/list", summary="分页查询{{ function_name }}", response_model=ResponseSchema[PageResultSchema[{{ class_name }}OutSchema]])
async def get_obj_list_controller(
    auth: Annotated[AuthSchema, Depends(AuthPermission(["{{ permission_prefix }}:query"]))],
    page: Annotated[PaginationQueryParam, Depends()],
    search: Annotated[{{ class_name }}QueryParam, Query()],
    db: Annotated[AsyncSession, Depends(db_getter)],
) -> JSONResponse:
    service = {{ class_name }}Service(auth, db)
    result_dict = await service.page(
        page_no=page.page_no,
        page_size=page.page_size,
        search=search,
        order_by=page.order_by,
    )
    return SuccessResponse(data=result_dict, msg="查询{{ function_name }}列表成功")


@{{ class_name }}Router.post("/create", status_code=status.HTTP_201_CREATED, summary="创建{{ function_name }}", response_model=ResponseSchema[{{ class_name }}OutSchema])
async def create_obj_controller(
    auth: Annotated[AuthSchema, Depends(AuthPermission(["{{ permission_prefix }}:create"]))],
    data: Annotated[{{ class_name }}CreateSchema, Body(description="创建参数")],
    db: Annotated[AsyncSession, Depends(db_getter)],
) -> JSONResponse:
    service = {{ class_name }}Service(auth, db)
    result_dict = await service.create(data=data)
    return SuccessResponse(data=result_dict, msg="创建{{ function_name }}成功")


@{{ class_name }}Router.put("/update/{id}", summary="修改{{ function_name }}", response_model=ResponseSchema[{{ class_name }}OutSchema])
async def update_obj_controller(
    auth: Annotated[AuthSchema, Depends(AuthPermission(["{{ permission_prefix }}:update"]))],
    id: Annotated[int, Path(description="{{ function_name }}ID")],
    data: Annotated[{{ class_name }}UpdateSchema, Body(description="修改参数")],
    db: Annotated[AsyncSession, Depends(db_getter)],
) -> JSONResponse:
    service = {{ class_name }}Service(auth, db)
    result_dict = await service.update(id=id, data=data)
    return SuccessResponse(data=result_dict, msg="修改{{ function_name }}成功")


@{{ class_name }}Router.delete("/delete", summary="删除{{ function_name }}", response_model=ResponseSchema[None])
async def delete_obj_controller(
    auth: Annotated[AuthSchema, Depends(AuthPermission(["{{ permission_prefix }}:delete"]))],
    ids: Annotated[list[int], Body(description="ID列表")],
    db: Annotated[AsyncSession, Depends(db_getter)],
) -> JSONResponse:
    service = {{ class_name }}Service(auth, db)
    await service.delete(ids=ids)
    return SuccessResponse(msg="删除{{ function_name }}成功")


@{{ class_name }}Router.patch("/status/batch", summary="批量修改{{ function_name }}状态", response_model=ResponseSchema[None])
async def batch_set_available_obj_controller(
    auth: Annotated[AuthSchema, Depends(AuthPermission(["{{ permission_prefix }}:patch"]))],
    data: Annotated[BatchSetAvailable, Body(description="状态设置")],
    db: Annotated[AsyncSession, Depends(db_getter)],
) -> JSONResponse:
    service = {{ class_name }}Service(auth, db)
    await service.set_available(data=data)
    return SuccessResponse(msg="批量修改{{ function_name }}状态成功")


@{{ class_name }}Router.post("/export", summary="导出{{ function_name }}")
async def export_obj_list_controller(
    auth: Annotated[AuthSchema, Depends(AuthPermission(["{{ permission_prefix }}:export"]))],
    search: Annotated[{{ class_name }}QueryParam, Query()],
    db: Annotated[AsyncSession, Depends(db_getter)],
) -> StreamingResponse:
    service = {{ class_name }}Service(auth, db)
    result_dict_list = await service.get_list(search=search)
    export_result = {{ class_name }}Service.batch_export(obj_list=[item.model_dump() for item in result_dict_list])

    return StreamResponse(
        data=bytes2file_response(export_result),
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": "attachment; filename={{ table_name }}.xlsx"},
    )


@{{ class_name }}Router.post("/import", summary="导入{{ function_name }}", response_model=ResponseSchema[str])
async def import_obj_list_controller(
    file: Annotated[UploadFile, File(description="导入文件")],
    auth: Annotated[AuthSchema, Depends(AuthPermission(["{{ permission_prefix }}:import"]))],
    db: Annotated[AsyncSession, Depends(db_getter)],
) -> JSONResponse:
    service = {{ class_name }}Service(auth, db)
    batch_import_result = await service.batch_import(file=file, update_support=True)
    return SuccessResponse(data=batch_import_result, msg="导入{{ function_name }}成功")


@{{ class_name }}Router.post("/download/template", summary="获取{{ function_name }}导入模板", dependencies=[Depends(AuthPermission(["{{ permission_prefix }}:download"]))])
async def export_obj_template_controller() -> StreamingResponse:
    import_template_result = {{ class_name }}Service.import_template_download()

    return StreamResponse(
        data=bytes2file_response(import_template_result),
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={
            "Content-Disposition": f"attachment; filename={urllib.parse.quote('{{ function_name }}导入模板.xlsx')}",
            "Access-Control-Expose-Headers": "Content-Disposition",
        },
    )