wren
104 lines
5.9 KiB
Python
104 lines
5.9 KiB
Python
"""评查点控制器。"""
|
|
|
|
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.evaluationPointDto import (
|
|
EvaluationPointCreateDTO,
|
|
EvaluationPointUpdateDTO,
|
|
)
|
|
from fastapi_modules.fastapi_leaudit.services.evaluationPointService import IEvaluationPointService
|
|
from fastapi_modules.fastapi_leaudit.services.impl.evaluationPointServiceImpl import EvaluationPointServiceImpl
|
|
from fastapi_modules.fastapi_leaudit.services.impl.permissionServiceImpl import PermissionServiceImpl
|
|
from fastapi_modules.fastapi_leaudit.services.permissionService import IPermissionService
|
|
|
|
|
|
class EvaluationPointController(BaseController):
|
|
"""评查点控制器。"""
|
|
|
|
_PERMISSIONS = {
|
|
"list": "evaluation_point:list:read",
|
|
"detail": "evaluation_point:detail:read",
|
|
"create": "evaluation_point:create:write",
|
|
"update": "evaluation_point:update:write",
|
|
"delete": "evaluation_point:delete:delete",
|
|
}
|
|
|
|
def __init__(self):
|
|
super().__init__(prefix="/v3/evaluation-points", tags=["评查点"])
|
|
self.PointService: IEvaluationPointService = EvaluationPointServiceImpl()
|
|
self.PermissionService: IPermissionService = PermissionServiceImpl()
|
|
|
|
@self.router.get("")
|
|
async def ListEvaluationPoints(
|
|
name: str | None = Query(None, description="名称模糊搜索"),
|
|
code: str | None = Query(None, description="编码模糊搜索"),
|
|
risk: str | None = Query(None, description="风险等级"),
|
|
is_enabled: bool | None = Query(None, description="是否启用"),
|
|
evaluation_point_groups_pid: int | None = Query(None, description="一级分组ID"),
|
|
evaluation_point_groups_id: int | None = Query(None, description="二级分组ID"),
|
|
document_attribute_type: str | None = Query(None, description="文档属性类型"),
|
|
area: str | None = Query(None, description="地区"),
|
|
page: int = Query(1, ge=1, description="页码"),
|
|
page_size: int = Query(20, ge=1, le=500, description="分页大小"),
|
|
payload: dict = Depends(verify_access_token),
|
|
):
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["list"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有评查点查看权限", "data": None})
|
|
data = await self.PointService.ListPoints(
|
|
name,
|
|
code,
|
|
risk,
|
|
is_enabled,
|
|
evaluation_point_groups_pid,
|
|
evaluation_point_groups_id,
|
|
document_attribute_type,
|
|
area,
|
|
page,
|
|
page_size,
|
|
)
|
|
return JSONResponse(status_code=200, content=data.model_dump())
|
|
|
|
@self.router.get("/attribute-types")
|
|
async def GetEvaluationPointAttributeTypes(payload: dict = Depends(verify_access_token)):
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["list"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有评查点查看权限", "data": None})
|
|
data = await self.PointService.GetAttributeTypes()
|
|
return JSONResponse(status_code=200, content=data.model_dump())
|
|
|
|
@self.router.get("/{PointId}")
|
|
async def GetEvaluationPoint(PointId: int, payload: dict = Depends(verify_access_token)):
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["detail"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有评查点查看权限", "data": None})
|
|
data = await self.PointService.GetPoint(PointId)
|
|
return JSONResponse(status_code=200, content=data.model_dump())
|
|
|
|
@self.router.post("")
|
|
async def CreateEvaluationPoint(body: EvaluationPointCreateDTO, payload: dict = Depends(verify_access_token)):
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["create"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有创建评查点权限", "data": None})
|
|
data = await self.PointService.CreatePoint(body)
|
|
return JSONResponse(status_code=200, content=data.model_dump())
|
|
|
|
@self.router.put("/{PointId}")
|
|
async def UpdateEvaluationPoint(PointId: int, body: EvaluationPointUpdateDTO, payload: dict = Depends(verify_access_token)):
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["update"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有更新评查点权限", "data": None})
|
|
data = await self.PointService.UpdatePoint(PointId, body)
|
|
return JSONResponse(status_code=200, content=data.model_dump())
|
|
|
|
@self.router.delete("/{PointId}")
|
|
async def DeleteEvaluationPoint(PointId: int, payload: dict = Depends(verify_access_token)):
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["delete"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有删除评查点权限", "data": None})
|
|
data = await self.PointService.DeletePoint(PointId)
|
|
return JSONResponse(status_code=200, content=data.model_dump())
|
|
|
|
async def _check_permission(self, user_id: int, permission_keys: list[str]) -> bool:
|
|
for permission_key in permission_keys:
|
|
if await self.PermissionService.CheckPermission(user_id, permission_key):
|
|
return True
|
|
return False
|