wren
46 lines
2.5 KiB
Python
46 lines
2.5 KiB
Python
"""规则配置页聚合控制器。"""
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import Depends
|
|
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.services.impl.permissionServiceImpl import PermissionServiceImpl
|
|
from fastapi_modules.fastapi_leaudit.services.impl.ruleConfigServiceImpl import RuleConfigServiceImpl
|
|
from fastapi_modules.fastapi_leaudit.services.permissionService import IPermissionService
|
|
from fastapi_modules.fastapi_leaudit.services.ruleConfigService import IRuleConfigService
|
|
|
|
|
|
class RuleConfigController(BaseController):
|
|
"""规则配置页聚合控制器。"""
|
|
|
|
def __init__(self):
|
|
super().__init__(prefix="/v3/rule-config-packs", tags=["规则配置"])
|
|
self.RuleConfigService: IRuleConfigService = RuleConfigServiceImpl()
|
|
self.PermissionService: IPermissionService = PermissionServiceImpl()
|
|
|
|
@self.router.get("")
|
|
async def ListRuleConfigPacks(payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""列出规则配置页 pack。"""
|
|
if not await self._check_permission(int(payload["user_id"])):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有规则配置查看权限", "data": None})
|
|
data = await self.RuleConfigService.ListPacks()
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "success", "data": [item.model_dump() for item in data]})
|
|
|
|
@self.router.get("/{PackId}")
|
|
async def GetRuleConfigPack(PackId: int, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""获取单个规则配置 pack。"""
|
|
if not await self._check_permission(int(payload["user_id"])):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有规则配置查看权限", "data": None})
|
|
data = await self.RuleConfigService.GetPack(PackId)
|
|
return JSONResponse(status_code=200, content={"code": 200, "message": "success", "data": data.model_dump()})
|
|
|
|
async def _check_permission(self, user_id: int) -> bool:
|
|
for permission_key in ("rules:list:read", "rules:content:read", "evaluation_group:list:read"):
|
|
if await self.PermissionService.CheckPermission(user_id, permission_key):
|
|
return True
|
|
return False
|