wren
142 lines
8.5 KiB
Python
142 lines
8.5 KiB
Python
"""RBAC 管理控制器。"""
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import Depends, Query
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from fastapi_common.fastapi_common_security.security import verify_access_token
|
|
from fastapi_common.fastapi_common_web.controller import BaseController
|
|
|
|
from fastapi_modules.fastapi_leaudit.domian.Dto.rbacAdminDto import RoleAccessSaveDTO, RoleCreateDTO, RolePermissionsBatchDTO, RoleRoutesUpdateDTO, RoleUpdateDTO, UserRolesAssignDTO
|
|
from fastapi_modules.fastapi_leaudit.services.impl.rbacAdminServiceImpl import RbacAdminServiceImpl
|
|
from fastapi_modules.fastapi_leaudit.services.rbacAdminService import IRbacAdminService
|
|
|
|
|
|
class RbacAdminController(BaseController):
|
|
"""RBAC 管理控制器。"""
|
|
|
|
def __init__(self):
|
|
super().__init__(prefix="", tags=["RBAC管理"])
|
|
self.RbacAdminService: IRbacAdminService = RbacAdminServiceImpl()
|
|
|
|
@self.router.get("/v3/rbac/roles")
|
|
async def GetRoles(
|
|
payload: dict[str, Any] = Depends(verify_access_token),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
role_key: str | None = Query(None),
|
|
role_name: str | None = Query(None),
|
|
include_system: bool = Query(True),
|
|
):
|
|
"""查询角色列表。"""
|
|
data = await self.RbacAdminService.ListRoles(int(payload["user_id"]), page, page_size, role_key, role_name, include_system)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "success", "data": data.model_dump()})
|
|
|
|
@self.router.post("/v3/rbac/roles")
|
|
async def CreateRole(Body: RoleCreateDTO, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""创建角色。"""
|
|
data = await self.RbacAdminService.CreateRole(int(payload["user_id"]), Body)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "角色创建成功", "data": data.model_dump()})
|
|
|
|
@self.router.put("/v3/rbac/roles/{RoleId}")
|
|
async def UpdateRole(RoleId: int, Body: RoleUpdateDTO, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""更新角色。"""
|
|
data = await self.RbacAdminService.UpdateRole(int(payload["user_id"]), RoleId, Body)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "角色更新成功", "data": data.model_dump()})
|
|
|
|
@self.router.delete("/v3/rbac/roles/{RoleId}")
|
|
async def DeleteRole(RoleId: int, force: bool = Query(False), payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""删除角色。"""
|
|
await self.RbacAdminService.DeleteRole(int(payload["user_id"]), RoleId, force)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "角色删除成功", "data": {}})
|
|
|
|
@self.router.get("/v3/rbac/users")
|
|
async def GetUsers(
|
|
payload: dict[str, Any] = Depends(verify_access_token),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
area: str | None = Query(None),
|
|
nick_name: str | None = Query(None),
|
|
):
|
|
"""查询用户列表。"""
|
|
data = await self.RbacAdminService.ListUsers(int(payload["user_id"]), page, page_size, area, nick_name)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "success", "data": data.model_dump()})
|
|
|
|
@self.router.get("/v3/rbac/roles/{RoleId}/users")
|
|
async def GetRoleUsers(
|
|
RoleId: int,
|
|
payload: dict[str, Any] = Depends(verify_access_token),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
area: str | None = Query(None),
|
|
username: str | None = Query(None),
|
|
):
|
|
"""查询指定角色下的用户列表。"""
|
|
data = await self.RbacAdminService.ListRoleUsers(int(payload["user_id"]), RoleId, page, page_size, area, username)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "success", "data": data.model_dump()})
|
|
|
|
@self.router.post("/v3/rbac/users/{UserId}/roles")
|
|
async def AssignUserRoles(UserId: int, Body: UserRolesAssignDTO, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""分配用户角色。"""
|
|
data = await self.RbacAdminService.AssignUserRoles(int(payload["user_id"]), UserId, Body.role_ids)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "角色分配成功", "data": data.model_dump()})
|
|
|
|
@self.router.delete("/v3/rbac/users/{UserId}/roles/{RoleId}")
|
|
async def RevokeUserRole(UserId: int, RoleId: int, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""移除用户角色。"""
|
|
await self.RbacAdminService.RevokeUserRole(int(payload["user_id"]), UserId, RoleId)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "角色移除成功", "data": {}})
|
|
|
|
@self.router.get("/v3/rbac/users/{UserId}/roles")
|
|
async def GetUserRoles(UserId: int, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""查询用户角色。"""
|
|
data = await self.RbacAdminService.GetUserRoles(int(payload["user_id"]), UserId)
|
|
return JSONResponse(status_code=200, content={"code": 200, "msg": "success", "data": data.model_dump()})
|
|
|
|
@self.router.get("/v3/routes")
|
|
async def GetAllRoutes(
|
|
payload: dict[str, Any] = Depends(verify_access_token),
|
|
format: str = Query("tree"),
|
|
include_hidden: bool = Query(False),
|
|
):
|
|
"""查询全部可管理路由。"""
|
|
data = await self.RbacAdminService.ListAllRoutes(int(payload["user_id"]), format, include_hidden)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "success", "data": [item.model_dump() for item in data]})
|
|
|
|
@self.router.get("/rbac/roles/{RoleId}/routes")
|
|
async def GetRoleRoutes(RoleId: int, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""查询角色路由授权。"""
|
|
data = await self.RbacAdminService.GetRoleRoutes(int(payload["user_id"]), RoleId)
|
|
return JSONResponse(status_code=200, content={"code": 200, "msg": "success", "data": data.model_dump()})
|
|
|
|
@self.router.put("/rbac/roles/{RoleId}/routes")
|
|
async def UpdateRoleRoutes(RoleId: int, Body: RoleRoutesUpdateDTO, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""更新角色路由授权。"""
|
|
data = await self.RbacAdminService.UpdateRoleRoutes(int(payload["user_id"]), RoleId, Body)
|
|
return JSONResponse(status_code=200, content={"code": 200, "msg": "success", "data": data.model_dump()})
|
|
|
|
@self.router.get("/v3/rbac/role-permissions")
|
|
async def GetRolePermissions(role_id: int = Query(...), payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""查询角色权限授权。"""
|
|
data = await self.RbacAdminService.GetRolePermissions(int(payload["user_id"]), role_id)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "success", "data": data.model_dump()})
|
|
|
|
@self.router.post("/v3/rbac/role-permissions")
|
|
async def SaveRolePermissions(Body: RolePermissionsBatchDTO, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""保存角色权限授权。"""
|
|
data = await self.RbacAdminService.SaveRolePermissions(int(payload["user_id"]), Body)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "权限分配成功", "data": data.model_dump()})
|
|
|
|
@self.router.post("/v3/rbac/roles/{RoleId}/access")
|
|
async def SaveRoleAccess(RoleId: int, Body: RoleAccessSaveDTO, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""原子保存角色菜单与接口权限。"""
|
|
data = await self.RbacAdminService.SaveRoleAccess(int(payload["user_id"]), RoleId, Body)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "角色权限保存成功", "data": data.model_dump()})
|
|
|
|
@self.router.get("/v3/routes/{RouteId}/permissions")
|
|
async def GetRoutePermissions(RouteId: int, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""查询路由关联权限。"""
|
|
data = await self.RbacAdminService.GetRoutePermissions(int(payload["user_id"]), RouteId)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "success", "data": data.model_dump()})
|