feat: add rbac-backed settings modules

2026-04-29 22:25:06 +08:00
parent b3ad4a6f33
commit 3a58f19d6c
23 changed files with 2979 additions and 7 deletions
@@ -84,7 +84,7 @@ class AuthController(BaseController):
            except LeauditException as e:
                logger.error(f"登录失败: {e.message}")
                return JSONResponse(
-                    status_code=e.statusCode,
+                    status_code=e.status.value,
                    content={"success": False, "message": e.message, "data": None},
                )
            except Exception as e:
@@ -106,7 +106,7 @@ class AuthController(BaseController):
            except LeauditException as e:
                logger.error(f"密码登录失败: {e.message}")
                return JSONResponse(
-                    status_code=e.statusCode,
+                    status_code=e.status.value,
                    content={"success": False, "message": e.message, "data": None},
                )
            except Exception as e:
@@ -0,0 +1,85 @@
+"""入口模块管理控制器。"""
+
+from fastapi import Depends, File, Query, UploadFile
+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.entryModuleDto import EntryModuleCreateDTO, EntryModuleUpdateDTO
+from fastapi_modules.fastapi_leaudit.services.entryModuleAdminService import IEntryModuleAdminService
+from fastapi_modules.fastapi_leaudit.services.impl.entryModuleAdminServiceImpl import EntryModuleAdminServiceImpl
+from fastapi_modules.fastapi_leaudit.services.impl.permissionServiceImpl import PermissionServiceImpl
+from fastapi_modules.fastapi_leaudit.services.permissionService import IPermissionService
+
+
+class EntryModuleController(BaseController):
+    """入口模块管理控制器。"""
+
+    def __init__(self):
+        super().__init__(prefix="/v3/entry-modules", tags=["入口模块管理"])
+        self.EntryModuleService: IEntryModuleAdminService = EntryModuleAdminServiceImpl()
+        self.PermissionService: IPermissionService = PermissionServiceImpl()
+
+        @self.router.get("")
+        async def GetEntryModules(
+            name: str | None = Query(None, description="模块名称模糊搜索"),
+            area: str | None = Query(None, description="地区筛选"),
+            page: int = Query(1, ge=1, description="页码"),
+            page_size: int = Query(10, ge=1, le=200, description="每页数量"),
+            payload: dict = Depends(verify_access_token),
+        ):
+            """查询入口模块列表。"""
+            if not await self.PermissionService.CheckPermission(int(payload["user_id"]), "entry_module:list:read"):
+                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有入口模块列表权限", "data": None})
+            data = await self.EntryModuleService.ListModules(Name=name, Area=area, Page=page, PageSize=page_size)
+            return JSONResponse(status_code=200, content={"code": 0, "msg": "success", "data": data.model_dump()})
+
+        @self.router.get("/{ModuleId}")
+        async def GetEntryModule(ModuleId: int, payload: dict = Depends(verify_access_token)):
+            """查询入口模块详情。"""
+            if not await self.PermissionService.CheckPermission(int(payload["user_id"]), "entry_module:detail:read"):
+                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有入口模块详情权限", "data": None})
+            data = await self.EntryModuleService.GetModule(ModuleId)
+            return JSONResponse(status_code=200, content={"code": 0, "msg": "success", "data": data.model_dump()})
+
+        @self.router.post("")
+        async def CreateEntryModule(Body: EntryModuleCreateDTO, payload: dict = Depends(verify_access_token)):
+            """创建入口模块。"""
+            if not await self.PermissionService.CheckPermission(int(payload["user_id"]), "entry_module:create:write"):
+                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有创建入口模块权限", "data": None})
+            data = await self.EntryModuleService.CreateModule(Body)
+            return JSONResponse(status_code=200, content={"code": 0, "msg": "success", "data": data.model_dump()})
+
+        @self.router.put("/{ModuleId}")
+        async def UpdateEntryModule(ModuleId: int, Body: EntryModuleUpdateDTO, payload: dict = Depends(verify_access_token)):
+            """更新入口模块。"""
+            if not await self.PermissionService.CheckPermission(int(payload["user_id"]), "entry_module:update:write"):
+                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有更新入口模块权限", "data": None})
+            data = await self.EntryModuleService.UpdateModule(ModuleId, Body)
+            return JSONResponse(status_code=200, content={"code": 0, "msg": "success", "data": data.model_dump()})
+
+        @self.router.delete("/{ModuleId}")
+        async def DeleteEntryModule(ModuleId: int, payload: dict = Depends(verify_access_token)):
+            """删除入口模块。"""
+            if not await self.PermissionService.CheckPermission(int(payload["user_id"]), "entry_module:delete:delete"):
+                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有删除入口模块权限", "data": None})
+            await self.EntryModuleService.DeleteModule(ModuleId)
+            return JSONResponse(status_code=200, content={"code": 0, "msg": "success", "data": {"message": "删除成功"}})
+
+        @self.router.post("/{ModuleId}/image")
+        async def UploadEntryModuleImage(
+            ModuleId: int,
+            file: UploadFile = File(..., description="入口模块图标"),
+            payload: dict = Depends(verify_access_token),
+        ):
+            """上传入口模块图标。"""
+            if not await self.PermissionService.CheckPermission(int(payload["user_id"]), "entry_module:image:write"):
+                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有上传入口模块图标权限", "data": None})
+            content = await file.read()
+            data = await self.EntryModuleService.UploadModuleImage(
+                ModuleId=ModuleId,
+                FileName=file.filename or f"entry_module_{ModuleId}.png",
+                ContentType=file.content_type or "application/octet-stream",
+                Content=content,
+            )
+            return JSONResponse(status_code=200, content={"code": 0, "msg": "success", "data": data.model_dump()})
@@ -0,0 +1,27 @@
+"""首页入口控制器。"""
+
+from typing import Any
+
+from fastapi import Depends
+
+from fastapi_common.fastapi_common_security.security import verify_access_token
+from fastapi_common.fastapi_common_web.controller import BaseController
+from fastapi_common.fastapi_common_web.domain.responses import Result
+
+from fastapi_modules.fastapi_leaudit.domian.vo.homeVo import HomeEntryModuleVO
+from fastapi_modules.fastapi_leaudit.services import IHomeService
+from fastapi_modules.fastapi_leaudit.services.impl.homeServiceImpl import HomeServiceImpl
+
+
+class HomeController(BaseController):
+    """首页入口控制器。"""
+
+    def __init__(self):
+        super().__init__(prefix="/home", tags=["首页"])
+        self.HomeService: IHomeService = HomeServiceImpl()
+
+        @self.router.get("/entry-modules", response_model=Result[list[HomeEntryModuleVO]])
+        async def GetEntryModules(payload: dict[str, Any] = Depends(verify_access_token)):
+            """获取当前用户可见的首页入口模块。"""
+            Data = await self.HomeService.GetEntryModules(UserId=int(payload["user_id"]))
+            return Result.success(data=Data)
@@ -0,0 +1,135 @@
+"""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 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.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()})
@@ -0,0 +1,27 @@
+"""RBAC 路由控制器。"""
+
+from typing import Any
+
+from fastapi import Depends
+
+from fastapi_common.fastapi_common_security.security import verify_access_token
+from fastapi_common.fastapi_common_web.controller import BaseController
+from fastapi_common.fastapi_common_web.domain.responses import Result
+
+from fastapi_modules.fastapi_leaudit.domian.vo.rbacVo import RbacUserRoutesVO
+from fastapi_modules.fastapi_leaudit.services import IRbacService
+from fastapi_modules.fastapi_leaudit.services.impl.rbacServiceImpl import RbacServiceImpl
+
+
+class RbacController(BaseController):
+    """RBAC 路由控制器。"""
+
+    def __init__(self):
+        super().__init__(prefix="/rbac", tags=["RBAC"])
+        self.RbacService: IRbacService = RbacServiceImpl()
+
+        @self.router.get("/user/routes", response_model=Result[RbacUserRoutesVO])
+        async def GetCurrentUserRoutes(payload: dict[str, Any] = Depends(verify_access_token)):
+            """获取当前登录用户可访问的前端路由树。"""
+            Data = await self.RbacService.GetCurrentUserRoutes(UserId=int(payload["user_id"]))
+            return Result.success(data=Data)