Skip to content

refactor: refactor permission handling to utilize PermissionConstants for improved clarity and maintainability #7002

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
shaohuzhang1 wants to merge 1 commit into from
+72 −57
Closed

refactor: refactor permission handling to utilize PermissionConstants for improved clarity and maintainability #7002

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 2 additions & 3 deletions apps/system_manage/serializers/user_group_resource_permission.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
from common.constants.cache_version import Cache_Version
from common.auth.constants.role_constants import RoleConstants
from common.constants.resource_permission_constants import ResourceAuthType, ResourcePermissionConstants
from common.database_model_manage.database_model_manage import DatabaseModelManage
from common.db.search import native_search, native_page_search, get_dynamics_model
from common.db.sql_execute import select_list
from common.exception.app_exception import AppApiException
Expand Down Expand Up @@ -373,7 +372,7 @@ def get_queryset(self, instance):
permission = instance.get("permission")
query_p_list = [None if p == "NOT_AUTH" else p for p in permission]

workspace_user_resource_permission_query_set = QuerySet(WorkspaceUserGroupResourcePermission).filter(
workspace_user_group_resource_permission_query_set = QuerySet(WorkspaceUserGroupResourcePermission).filter(
workspace_id=self.data.get("workspace_id"),
auth_target_type=self.data.get("auth_target_type"),
target=self.data.get("target"),
Expand All @@ -389,7 +388,7 @@ def get_queryset(self, instance):
else:
user_query_set = user_query_set.filter(permission__in=query_p_list)
return {
"workspace_user_resource_permission_query_set": workspace_user_resource_permission_query_set,
"workspace_user_group_resource_permission_query_set": workspace_user_group_resource_permission_query_set,
"user_query_set": user_query_set,
}

Expand Down
124 changes: 70 additions & 54 deletions apps/system_manage/views/resource_mapping.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
# coding=utf-8
"""
@project: MaxKB
@Author：虎虎
@file： resource_mapping.py
@date：2025/12/25 15:28
@desc:
@project: MaxKB
@Author：虎虎
@file： resource_mapping.py
@date：2025/12/25 15:28
@desc:
"""

from django.utils.translation import gettext_lazy as _
Expand All @@ -16,11 +16,9 @@
from common.auth import TokenAuth
from common.auth.authentication import has_permissions
from common.auth.constants.compare_constants import CompareConstants
from common.auth.constants.group_constants import Group
from common.auth.constants.operate_constants import Operate
from common.auth.constants.permission_constants import PermissionConstants
from common.auth.constants.role_constants import RoleConstants
from common.auth.struct.aggregate_permission import ViewPermission
from common.auth.struct.permission import Permission
from system_manage.api.resource_mapping import ResourceMappingAPI
from system_manage.serializers.resource_mapping_serializers import ResourceMappingSerializer, MappingResourceSerializer

Expand All @@ -29,65 +27,83 @@ class ResourceMappingView(APIView):
authentication_classes = [TokenAuth]

@extend_schema(
methods=['GET'],
description=_('Retrieve the pagination list of resource relationships'),
operation_id=_('Retrieve the pagination list of resource relationships'), # type: ignore
methods=["GET"],
description=_("Retrieve the pagination list of resource relationships"),
operation_id=_("Retrieve the pagination list of resource relationships"), # type: ignore
responses=ResourceMappingAPI.get_response(),
parameters=ResourceMappingAPI.get_parameters(),
tags=[_('Resources mapping')] # type: ignore
tags=[_("Resources mapping")], # type: ignore
)
@has_permissions(
lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.RELATE_VIEW,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}:ROLE/WORKSPACE_MANAGE"),
lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.RELATE_VIEW,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource')}/{kwargs.get('resource_id')}"),
ViewPermission([RoleConstants.USER.get_workspace_role()],
[lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.SELF,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource')}/{kwargs.get('resource_id')}")],
compare=CompareConstants.AND),
RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
lambda r, kwargs: PermissionConstants[
f"{kwargs.get('resource')}_RELATE_RESOURCE_VIEW"
].get_workspace_permission_workspace_manage_role()(r, kwargs),
lambda r, kwargs: PermissionConstants[
f"{kwargs.get('resource')}_RELATE_RESOURCE_VIEW"
]._build_workspace_permission(resource_id_key="resource_id")(r, kwargs),
ViewPermission(
[RoleConstants.USER.get_workspace_role()],
[
lambda r, kwargs: PermissionConstants[kwargs.get("resource")]._build_workspace_permission(
resource_id_key="resource_id"
)(r, kwargs)
],
compare=CompareConstants.AND,
),
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(),
)
def get(self, request: Request, workspace_id: str, resource: str, resource_id: str, current_page, page_size):
return result.success(ResourceMappingSerializer({
'resource': resource,
'resource_id': resource_id,
'resource_name': request.query_params.get('resource_name'),
'user_name': request.query_params.get('user_name'),
'source_type': request.query_params.getlist('source_type[]'),
}).page(current_page, page_size))
return result.success(
ResourceMappingSerializer(
{
"resource": resource,
"resource_id": resource_id,
"resource_name": request.query_params.get("resource_name"),
"user_name": request.query_params.get("user_name"),
"source_type": request.query_params.getlist("source_type[]"),
}
).page(current_page, page_size)
)


class MappingResourceView(APIView):
authentication_classes = [TokenAuth]

@extend_schema(
methods=['GET'],
description=_('Retrieve the pagination list of resource relationships'),
operation_id=_('Retrieve the pagination list of resource relationships'), # type: ignore
methods=["GET"],
description=_("Retrieve the pagination list of resource relationships"),
operation_id=_("Retrieve the pagination list of resource relationships"), # type: ignore
responses=ResourceMappingAPI.get_response(),
parameters=ResourceMappingAPI.get_parameters(),
tags=[_('Mapping Resource')] # type: ignore
tags=[_("Mapping Resource")], # type: ignore
)
@has_permissions(
lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.RELATE_VIEW,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}:ROLE/WORKSPACE_MANAGE"),
lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.RELATE_VIEW,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource')}/{kwargs.get('resource_id')}"),
ViewPermission([RoleConstants.USER.get_workspace_role()],
[lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.SELF,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource')}/{kwargs.get('resource_id')}")],
compare=CompareConstants.AND),
RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
lambda r, kwargs: PermissionConstants[
f"{kwargs.get('resource')}_RELATE_RESOURCE_VIEW"
].get_workspace_permission_workspace_manage_role()(r, kwargs),
lambda r, kwargs: PermissionConstants[
f"{kwargs.get('resource')}_RELATE_RESOURCE_VIEW"
]._build_workspace_permission(resource_id_key="resource_id")(r, kwargs),
ViewPermission(
[RoleConstants.USER.get_workspace_role()],
[
lambda r, kwargs: PermissionConstants[kwargs.get("resource")]._build_workspace_permission(
resource_id_key="resource_id"
)(r, kwargs)
],
compare=CompareConstants.AND,
),
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(),
)
def get(self, request: Request, workspace_id: str, resource: str, resource_id: str, current_page, page_size):
return result.success(MappingResourceSerializer({
'resource': resource,
'resource_id': resource_id,
'resource_name': request.query_params.get('resource_name'),
'user_name': request.query_params.get('user_name'),
'target_type': request.query_params.getlist('target_type[]'),
}).page(current_page, page_size))
return result.success(
MappingResourceSerializer(
{
"resource": resource,
"resource_id": resource_id,
"resource_name": request.query_params.get("resource_name"),
"user_name": request.query_params.get("user_name"),
"target_type": request.query_params.getlist("target_type[]"),
}
).page(current_page, page_size)
)
Loading