# _*_ coding : UTF-8 _*_ # @Time : 2025/01/20 22:41 # @UpdateTime : 2025/01/20 22:41 # @Author : sonder # @File : role.py # @Software : PyCharm # @Comment : 本程序 from typing import Optional from fastapi import APIRouter, Depends, Path, Query, Request from fastapi.responses import JSONResponse from annotation.auth import Auth, hasAuth, hasAdmin from annotation.log import Log from config.constant import BusinessType, RedisKeyConfig from controller.login import LoginController from models import Role, Permission, RolePermission, Department from schemas.common import BaseResponse, DeleteListParams from schemas.role import AddRoleParams, AddRolePermissionParams, GetRolePermissionInfoResponse, \ GetRolePermissionListResponse from utils.common import filterKeyValues from utils.response import Response roleAPI = APIRouter( prefix="/role" ) @roleAPI.post("/add", response_model=BaseResponse, response_class=JSONResponse, summary="新增角色") @Log(title="新增角色", business_type=BusinessType.INSERT) @Auth(permission_list=["role:btn:add"]) async def add_role(request: Request, params: AddRoleParams, current_user: dict = Depends(LoginController.get_current_user)): if await Role.get_or_none(code=params.code, department_id=params.department_id, del_flag=1): return Response.error(msg="角色编码已存在！") sub_departments = current_user.get("sub_departments") if params.department_id not in sub_departments: return Response.error(msg="新增失败,无权限！") department = await Department.get_or_none(id=params.department_id, del_flag=1) if department: role = await Role.create( code=params.code, name=params.name, description=params.description, status=params.status, department_id=department.id, ) else: role = await Role.create( code=params.code, name=params.name, status=params.status, description=params.description, department_id=None, ) if role: # 更新用户信息缓存 userInfos = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_INFO.key}*') if userInfos: await request.app.state.redis.delete(*userInfos) # 更新用户路由缓存 userRoutes = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_ROUTES.key}*') if userRoutes: await request.app.state.redis.delete(*userRoutes) return Response.success(msg="新增角色成功！") return Response.error(msg="新增角色失败！") @roleAPI.delete("/delete/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="删除角色") @roleAPI.post("/delete/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="删除角色") @Log(title="删除角色", business_type=BusinessType.DELETE) @Auth(permission_list=["role:btn:delete"]) async def delete_role(request: Request, id: int = Path(..., description="角色ID"), current_user: dict = Depends(LoginController.get_current_user)): sub_departments = current_user.get("sub_departments") if role := await Role.get_or_none(id=id, department__id__in=sub_departments, del_flag=1): # 移除相应角色权限 await RolePermission.filter(role_id=role.id, del_flag=1).update(del_flag=0) role.del_flag = 0 await role.save() # 更新用户信息缓存 userInfos = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_INFO.key}*') if userInfos: await request.app.state.redis.delete(*userInfos) # 更新用户路由缓存 userRoutes = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_ROUTES.key}*') if userRoutes: await request.app.state.redis.delete(*userRoutes) return Response.success(msg="删除角色成功！") return Response.error(msg="删除角色失败！") @roleAPI.delete("/deleteList", response_model=BaseResponse, response_class=JSONResponse, summary="批量删除角色") @roleAPI.post("/deleteList", response_model=BaseResponse, response_class=JSONResponse, summary="批量删除角色") @Log(title="批量删除角色", business_type=BusinessType.DELETE) @Auth(permission_list=["role:btn:delete"]) async def delete_role_list(request: Request, params: DeleteListParams, current_user: dict = Depends(LoginController.get_current_user)): sub_departments = current_user.get("sub_departments") for id in set(params.ids): if role := await Role.get_or_none(id=id, del_flag=1, department__id__in=sub_departments): # 移除相应角色权限 await RolePermission.filter(role_id=role.id, del_flag=1).update(del_flag=0) role.del_flag = 0 await role.save() # 更新用户信息缓存 userInfos = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_INFO.key}*') if userInfos: await request.app.state.redis.delete(*userInfos) # 更新用户路由缓存 userRoutes = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_ROUTES.key}*') if userRoutes: await request.app.state.redis.delete(*userRoutes) return Response.success(msg="批量删除角色成功！") @roleAPI.put("/update/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="修改角色") @roleAPI.post("/update/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="修改角色") @Log(title="修改角色", business_type=BusinessType.UPDATE) @Auth(permission_list=["role:btn:update"]) async def update_role(request: Request, params: AddRoleParams, id: str = Path(..., description="角色ID"), current_user: dict = Depends(LoginController.get_current_user)): sub_departments = current_user.get("sub_departments") if params.department_id not in sub_departments: return Response.error(msg="修改失败,无权限！") if role := await Role.get_or_none(id=id, del_flag=1): role.code = params.code role.name = params.name role.description = params.description department = await Department.get_or_none(id=params.department_id, del_flag=1) role.status = params.status if department: role.department_id = department.id else: role.department_id = None await role.save() # 更新用户信息缓存 userInfos = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_INFO.key}*') if userInfos: await request.app.state.redis.delete(*userInfos) # 更新用户路由缓存 userRoutes = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_ROUTES.key}*') if userRoutes: await request.app.state.redis.delete(*userRoutes) return Response.success(msg="修改角色成功！") return Response.error(msg="修改角色失败！") @roleAPI.get("/info/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="查询角色详情") @Log(title="查询角色详情", business_type=BusinessType.SELECT) @Auth(permission_list=["role:btn:info"]) async def get_role_info(request: Request, id: int = Path(..., description="角色ID"), current_user: dict = Depends(LoginController.get_current_user)): sub_departments = current_user.get("sub_departments") if role := await Role.get_or_none(id=id, del_flag=1, department__id__in=sub_departments).values( id="id", create_by="create_by", create_time="create_time", update_by="update_by", update_time="update_time", code="code", name="name", status="status", description="description", department_id="department_id", department_name="department__name", department_principal="department__principal", department_phone="department__phone", department_email="department__email", ): return Response.success(data=role) return Response.error(msg="查询角色详情失败！") @roleAPI.get("/list", response_model=BaseResponse, response_class=JSONResponse, summary="查询角色列表") @Log(title="查询角色列表", business_type=BusinessType.SELECT) @Auth(permission_list=["role:btn:list"]) async def get_role_list( request: Request, page: int = Query(1, description="页码"), pageSize: int = Query(10, description="每页数量"), name: Optional[str] = Query(None, description="角色名称"), code: Optional[str] = Query(None, description="角色编码"), description: Optional[str] = Query(None, description="角色描述"), department_id: Optional[str] = Query(None, description="所属部门ID"), status: Optional[int] = Query(None, description="状态"), current_user: dict = Depends(LoginController.get_current_user) ): filterArgs = { f'{k}__icontains': v for k, v in { "name": name, "code": code, "description": description, "status": status }.items() if v } if await hasAuth(request, "role:btn:admin"): if not department_id: filterArgs["department__id__in"] = current_user.get("sub_departments") else: filterArgs["department__id"] = current_user.get("department_id") else: if department_id: filterArgs["department__id"] = department_id else: filterArgs["department__id"] = current_user.get("department_id") total = await Role.filter(**filterArgs, del_flag=1).count() data = await Role.filter(**filterArgs, del_flag=1).offset( (page - 1) * pageSize).limit( pageSize).values( id="id", create_by="create_by", create_time="create_time", update_by="update_by", update_time="update_time", code="code", name="name", status="status", description="description", department_id="department__id", department_name="department__name", department_principal="department__principal", department_phone="department__phone", department_email="department__email", ) return Response.success(data={ "result": data, "total": total, "page": page, "pageSize": pageSize }) @roleAPI.post("/addPermission", response_model=BaseResponse, response_class=JSONResponse, summary="新增角色权限") @Log(title="新增角色权限", business_type=BusinessType.INSERT) @Auth(permission_list=["role:btn:addPermission"]) async def add_role_permission(request: Request, params: AddRolePermissionParams, id: str = Path(..., description="角色ID"), current_user: dict = Depends(LoginController.get_current_user)): sub_departments = current_user.get("sub_departments") if await hasAdmin(request, current_user.get("department_id")): department_permissions = await Permission.filter(del_flag=1).values("id") else: department_permissions = await Permission.filter(is_admin=False, del_flag=1).values("id") department_permissions = filterKeyValues(department_permissions, "id") if role := await Role.get_or_none(id=id, del_flag=1, department__id__in=sub_departments): # 已有角色权限 rolePermissions = await RolePermission.filter(role_id=id, del_flag=1).values("permission_id") rolePermissions = await filterKeyValues(rolePermissions, "permission_id") # 利用集合筛选出角色权限中不存在的权限 add_list = set(params.permission_ids).difference(set(rolePermissions)) # 循环添加角色权限 for item in add_list: if item not in department_permissions: continue permission = await Permission.get_or_none(id=item, del_flag=1) if permission: await RolePermission.create( role_id=role.id, permission_id=permission.id ) # 更新用户信息缓存 userInfos = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_INFO.key}*') if userInfos: await request.app.state.redis.delete(*userInfos) # 更新用户路由缓存 userRoutes = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_ROUTES.key}*') if userRoutes: await request.app.state.redis.delete(*userRoutes) return Response.success(msg="新增角色权限成功！") return Response.error(msg="新增角色权限失败！") @roleAPI.delete("/deletePermission/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="删除角色权限") @roleAPI.post("/deletePermission/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="删除角色权限") @Log(title="删除角色权限", business_type=BusinessType.DELETE) @Auth(permission_list=["role:btn:deletePermission"]) async def delete_role_permission(request: Request, id: int = Path(..., description="角色权限ID"), current_user: dict = Depends(LoginController.get_current_user)): sub_departments = current_user.get("sub_departments") if rolePermission := await RolePermission.get_or_none(id=id, del_flag=1, role__department__id__in=sub_departments): rolePermission.del_flag = 0 await rolePermission.save() # 更新用户信息缓存 userInfos = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_INFO.key}*') if userInfos: await request.app.state.redis.delete(*userInfos) # 更新用户路由缓存 userRoutes = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_ROUTES.key}*') if userRoutes: await request.app.state.redis.delete(*userRoutes) return Response.success(msg="删除角色权限成功！") return Response.error(msg="删除角色权限失败！") @roleAPI.put("/updatePermission/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="修改角色权限") @roleAPI.post("/updatePermission/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="修改角色权限") @Log(title="修改角色权限", business_type=BusinessType.UPDATE) @Auth(permission_list=["role:btn:updatePermission"]) async def update_role_permission(request: Request, params: AddRolePermissionParams, id: str = Path(..., description="角色ID"), current_user: dict = Depends(LoginController.get_current_user)): sub_departments = current_user.get("sub_departments") if await hasAdmin(request, current_user.get("department_id")): department_permissions = await Permission.filter(del_flag=1).values("id") else: department_permissions = await Permission.filter(is_admin=False, del_flag=1).values("id") department_permissions = await filterKeyValues(department_permissions, key="id", convert_type=str) if role := await Role.get_or_none(id=id, del_flag=1, department__id__in=sub_departments): # 已有角色权限 rolePermissions = await RolePermission.filter(role_id=role.id, del_flag=1).values("permission_id") rolePermissions = await filterKeyValues(rolePermissions, key="permission_id", convert_type=str) # 利用集合筛选出角色权限中不存在的权限 delete_list = set(rolePermissions).difference(set(params.permission_ids)) # 利用集合筛选出角色权限中新增的权限 add_list = set(params.permission_ids).difference(set(rolePermissions)) # 循环删除角色权限 for item in delete_list: await RolePermission.filter(role_id=id, permission_id=item, del_flag=1).update(del_flag=0) # 循环添加角色权限 for item in add_list: if item not in department_permissions: continue await RolePermission.create(role_id=id, permission_id=item) # 更新用户信息缓存 userInfos = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_INFO.key}*') if userInfos: await request.app.state.redis.delete(*userInfos) # 更新用户路由缓存 userRoutes = await request.app.state.redis.keys(f'{RedisKeyConfig.USER_ROUTES.key}*') if userRoutes: await request.app.state.redis.delete(*userRoutes) return Response.success(msg="修改角色权限成功！") return Response.error(msg="修改角色权限失败！") @roleAPI.get("/permissionInfo/{id}", response_model=GetRolePermissionInfoResponse, response_class=JSONResponse, summary="获取角色权限信息") @Log(title="获取角色权限信息", business_type=BusinessType.SELECT) @Auth(permission_list=["role:btn:permissionInfo"]) async def get_role_permission_info(request: Request, id: int = Path(..., description="角色权限ID"), current_user: dict = Depends(LoginController.get_current_user)): sub_departments = current_user.get("sub_departments") if rolePermission := await RolePermission.get_or_none(id=id, del_flag=1, role__department__id__in=sub_departments): data = await rolePermission.first().values( id="id", create_by="create_by", create_time="create_time", update_by="update_by", update_time="update_time", role_id="role__id", role_name="role__name", role_code="role__code", permission_id="permission__id", permission_name="permission__title", permission_auth="permission__auths", permission_type="permission__menu_type" ) return Response.success(data=data) return Response.error(msg="获取角色权限信息失败！") @roleAPI.get("/permissionList/{id}", response_model=GetRolePermissionListResponse, response_class=JSONResponse, summary="获取角色权限列表") @Log(title="获取角色权限列表", business_type=BusinessType.SELECT) @Auth(permission_list=["role:btn:permissionList"]) async def get_role_permission_list(request: Request, id: str = Path(..., description="角色ID"), current_user: dict = Depends(LoginController.get_current_user)): sub_departments = current_user.get("sub_departments") total = await RolePermission.filter(role_id=id, role__department__id__in=sub_departments, del_flag=1).count() data = await RolePermission.filter(role_id=id, role__department__id__in=sub_departments, del_flag=1).values( id="id", create_by="create_by", create_time="create_time", update_by="update_by", update_time="update_time", role_id="role__id", role_name="role__name", role_code="role__code", permission_id="permission__id", permission_parent_id="permission__parent_id", permission_name="permission__title", permission_auth="permission__auths", permission_type="permission__menu_type" ) return Response.success(data={ "result": data, "total": total, "page": 1, "pageSize": 9999 })