from typing import Annotated from fastapi import APIRouter, Body, Depends, Path from fastapi.responses import JSONResponse, StreamingResponse from app.api.v1.module_system.auth.schema import AuthSchema from app.common.response import ResponseSchema, StreamResponse, SuccessResponse from app.core.base_params import PaginationQueryParam from app.core.base_schema import BatchSetAvailable from app.core.dependencies import AuthPermission from app.core.logger import log from app.core.router_class import OperationLogRoute from app.utils.common_util import bytes2file_response from .schema import ( RoleCreateSchema, RoleOutSchema, RolePermissionSettingSchema, RoleQueryParam, RoleUpdateSchema, ) from .service import RoleService RoleRouter = APIRouter(route_class=OperationLogRoute, prefix="/role", tags=["角色管理"]) @RoleRouter.get( "/list", summary="查询角色", description="查询角色", response_model=ResponseSchema[list[RoleOutSchema]], ) async def get_obj_list_controller( page: Annotated[PaginationQueryParam, Depends()], search: Annotated[RoleQueryParam, Depends()], auth: Annotated[AuthSchema, Depends(AuthPermission(["module_system:role:query"]))], ) -> JSONResponse: """ 查询角色 参数: - page (PaginationQueryParam): 分页查询参数模型 - search (RoleQueryParam): 查询参数模型 - auth (AuthSchema): 认证信息模型 返回: - JSONResponse: 分页查询结果JSON响应 """ order_by = [{"order": "asc"}] if page.order_by: order_by = page.order_by result_dict = await RoleService.get_role_page_service( auth=auth, page_no=page.page_no, page_size=page.page_size, search=search, order_by=order_by, ) log.info("查询角色成功") return SuccessResponse(data=result_dict, msg="查询角色成功") @RoleRouter.get( "/detail/{id}", summary="查询角色详情", description="查询角色详情", response_model=ResponseSchema[RoleOutSchema], ) async def get_obj_detail_controller( id: Annotated[int, Path(description="角色ID")], auth: Annotated[AuthSchema, Depends(AuthPermission(["module_system:role:detail"]))], ) -> JSONResponse: """ 查询角色详情 参数: - id (int): 角色ID - auth (AuthSchema): 认证信息模型 返回: - JSONResponse: 角色详情JSON响应 """ result_dict = await RoleService.get_role_detail_service(id=id, auth=auth) log.info(f"获取角色详情成功 {id}") return SuccessResponse(data=result_dict, msg="获取角色详情成功") @RoleRouter.post( "/create", summary="创建角色", description="创建角色", response_model=ResponseSchema[RoleOutSchema], ) async def create_obj_controller( data: RoleCreateSchema, auth: Annotated[AuthSchema, Depends(AuthPermission(["module_system:role:create"]))], ) -> JSONResponse: """ 创建角色 参数: - data (RoleCreateSchema): 创建角色模型 - auth (AuthSchema): 认证信息模型 返回: - JSONResponse: 创建角色JSON响应 """ result_dict = await RoleService.create_role_service(data=data, auth=auth) log.info(f"创建角色成功: {result_dict}") return SuccessResponse(data=result_dict, msg="创建角色成功") @RoleRouter.put( "/update/{id}", summary="修改角色", description="修改角色", response_model=ResponseSchema[RoleOutSchema], ) async def update_obj_controller( data: RoleUpdateSchema, id: Annotated[int, Path(description="角色ID")], auth: Annotated[AuthSchema, Depends(AuthPermission(["module_system:role:update"]))], ) -> JSONResponse: """ 修改角色 参数: - data (RoleUpdateSchema): 修改角色模型 - id (int): 角色ID - auth (AuthSchema): 认证信息模型 返回: - JSONResponse: 修改角色JSON响应 """ result_dict = await RoleService.update_role_service(id=id, data=data, auth=auth) log.info(f"修改角色成功: {result_dict}") return SuccessResponse(data=result_dict, msg="修改角色成功") @RoleRouter.delete( "/delete", summary="删除角色", description="删除角色", response_model=ResponseSchema[None], ) async def delete_obj_controller( ids: Annotated[list[int], Body(description="ID列表")], auth: Annotated[AuthSchema, Depends(AuthPermission(["module_system:role:delete"]))], ) -> JSONResponse: """ 删除角色 参数: - ids (list[int]): ID列表 - auth (AuthSchema): 认证信息模型 返回: - JSONResponse: 删除角色JSON响应 """ await RoleService.delete_role_service(ids=ids, auth=auth) log.info(f"删除角色成功: {ids}") return SuccessResponse(msg="删除角色成功") @RoleRouter.patch( "/available/setting", summary="批量修改角色状态", description="批量修改角色状态", response_model=ResponseSchema[None], ) async def batch_set_available_obj_controller( data: BatchSetAvailable, auth: Annotated[AuthSchema, Depends(AuthPermission(["module_system:role:patch"]))], ) -> JSONResponse: """ 批量修改角色状态 参数: - data (BatchSetAvailable): 批量修改角色状态模型 - auth (AuthSchema): 认证信息模型 返回: - JSONResponse: 批量修改角色状态JSON响应 """ await RoleService.set_role_available_service(data=data, auth=auth) log.info(f"批量修改角色状态成功: {data.ids}") return SuccessResponse(msg="批量修改角色状态成功") @RoleRouter.patch( "/permission/setting", summary="角色授权", description="角色授权", response_model=ResponseSchema[None], ) async def set_role_permission_controller( data: RolePermissionSettingSchema, auth: Annotated[AuthSchema, Depends(AuthPermission(["module_system:role:permission"]))], ) -> JSONResponse: """ 角色授权 参数: - data (RolePermissionSettingSchema): 角色授权模型 - auth (AuthSchema): 认证信息模型 返回: - JSONResponse: 角色授权JSON响应 """ await RoleService.set_role_permission_service(data=data, auth=auth) log.info(f"设置角色权限成功: {data}") return SuccessResponse(msg="授权角色成功") @RoleRouter.post( "/export", summary="导出角色", description="导出角色", response_model=ResponseSchema[None], ) async def export_obj_list_controller( search: Annotated[RoleQueryParam, Depends()], auth: Annotated[AuthSchema, Depends(AuthPermission(["module_system:role:export"]))], ) -> StreamingResponse: """ 导出角色 参数: - search (RoleQueryParam): 查询参数模型 - auth (AuthSchema): 认证信息模型 返回: - StreamingResponse: 导出角色流响应 """ role_query_result = await RoleService.get_role_list_service(search=search, auth=auth) role_export_result = await RoleService.export_role_list_service(role_list=role_query_result) log.info("导出角色成功") return StreamResponse( data=bytes2file_response(role_export_result), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": "attachment; filename=role.xlsx"}, )