# -*- coding:utf-8 -*- """ @Created on : 2023/6/30 15:17 @Author: hxl @Des: """ from fastapi import APIRouter, Request, Depends from pydantic import BaseModel from helper import respond_to, login_required from models import Role, Archive, ModuleMenu, User router = APIRouter(prefix='/role',dependencies=[Depends(login_required)]) # dependencies=[Depends(login_required)] class UpdateRoleAuth(BaseModel): name:str grade:int = 0 portion:dict masks:dict comment:str | None class RoleDeleteRequest(BaseModel): role_ids: list @router.get('', summary='角色列表') async def index(request:Request, page_no: int = 1, page_size: int = 20): """ 获取角色列表 :return: """ role_id = request.state.current_user.role_id current_role = await Role.get(id=role_id) offset = (page_no - 1) * page_size query = Role.filter(grade__lte=current_role.grade) count = await query.count() roles = await query.offset(offset).limit(page_size).all() return respond_to(data=dict(count=count, roles= roles)) @router.get('/get_role_list_all',summary='获取角色名称列表') async def get_role_list_all(id:str): """ 获取角色名称列表,按当前登录角色的等级，返回小于该等级的角色列表 :return: """ user_obj = await User.get_or_none(id = id) role_user_obj = await Role.get_or_none(id=user_obj.role_id) roles = await Role.filter(grade__lte=role_user_obj.grade).values() # 等级名称 grade_name ={ 0:'普通用户', 50:'管理员', 75:'平台管理员', 100:'超级管理员' } roles_list = [] for item in roles : item['grade_name'] = grade_name[item.get("grade")] roles_list.append(item) return respond_to(data=roles_list) @router.get('/get_module_menu_all',summary='获取模块菜单列表') async def get_module_menu_all(): """ 获取模块菜单列表 :return: """ archives = await Archive.all() modules = await ModuleMenu.all() # 中控权限 portion = list(filter( lambda item : item.id>2000,modules)) # 柜体权限 masks = list(filter(lambda item : item.id<2000,modules)) data={ "archives":archives, "portion":portion, 'masks':masks } return respond_to(data=data) @router.get('/get_role_module_menu',summary='获取当前角色的权限模块') async def get_role_module_menu(role_id:str): """ 获取当前角色的权限模块 :param id: str role_id :return: """ print('ddddddddddddddddddddddddd::',id) archives = await Archive.all() modules = await ModuleMenu.all() # user_obj = await User.get_or_none(id=id) # role_obj = await Role.get_or_none(id = user_obj.role_id) role_obj = await Role.get_or_none(id=role_id) # 中控所有权限 portion_all = list(filter( lambda item : item.id>2000,modules)) # 柜体所有权限 masks_all = list(filter(lambda item : item.id<2000,modules)) # # 角色的中控权限 protion = role_obj.portion # # 角色的柜体权限 masks = role_obj.masks # protion_keys = [list(d.keys()) for d in protion] # masks_keys = [list(d.keys()) for d in masks] # 构造返回给前端"portion": {"<试剂类型ID>": {"<模块ID>": "<模块值>", ...}, "<试剂类型ID>": {"<模块ID>": "<模块值>", ...}}, # print('ddddddddddddddddddddddddddddddddd::',protion) # portion_list = [] # for archive in archives: # portion_name = {archive.id:list(filter(lambda item: str(item.id) in protion.get(archive.id).keys(), modules)) if protion.get(archive.id) else {}} # portion_list.append(portion_name) # # portion_name = list(filter(lambda item: str(item.id) in protion.keys(), modules)) if protion else {} # masks_name = list(filter(lambda item: str(item.id) in masks.keys(), modules)) if masks else {} data = { "name":role_obj.name, "id":role_obj.id, "protion":portion_all, "masks":masks_all, "comment":role_obj.comment, 'grade':role_obj.grade, "params":{ "archives":archives, "protion":protion, "masks":masks } } return respond_to(data=data) @router.post('/create_role_module_menu',summary='新增角色权限') async def create_role_module_menu(port:UpdateRoleAuth): """ 新增角色权限 :param post: 新增角色信息的请求体 :return: """ archives = await Archive.all() role_obj = await Role.create(**port.dict()) # # 角色的中控权限 protion = role_obj.portion # # 角色的柜体权限 masks = role_obj.masks data = { "name": role_obj.name, "id": role_obj.id, "archives": archives, "protion": protion, "masks": masks, } return respond_to(data=data) @router.put('/update_role_module_menu/{id}',summary='编辑角色权限') async def update_role_module_menu(id:str,port:UpdateRoleAuth): """ :param id: str role_id :param post: 编辑角色信息的请求体 :return: """ archives = await Archive.all() role_origin = await Role.get_or_none(id=id) role_obj = await role_origin.update_from_dict(port.dict()) await role_obj.save() # # 角色的中控权限 protion = role_obj.portion # # 角色的柜体权限 masks = role_obj.masks data = { "name": role_obj.name, "id": role_obj.id, "archives": archives, "protion": protion, "masks": masks, } return respond_to(data=data) @router.delete('/delete_role',summary='删除角色') async def delete_role(body: RoleDeleteRequest): """ 删除角色，如果该角色有绑定用户，则不允许删除 :param id :return """ for role_id in body.role_ids: user_count = await User.filter(role_id=role_id).count() role_obj = await Role.get_or_none(id=role_id) if user_count != 0 : return respond_to(code=400,desc=f'删除失败，{role_obj.name} 角色已被其他用户绑定') await role_obj.filter(id__in=body.role_ids).delete() return respond_to(desc='删除角色成功')