diff --git a/apps/perms/api/application/application_permission.py b/apps/perms/api/application/application_permission.py index 44e11b54e..0ddec2aa6 100644 --- a/apps/perms/api/application/application_permission.py +++ b/apps/perms/api/application/application_permission.py @@ -1,12 +1,12 @@ # -*- coding: utf-8 -*- # -from common.permissions import IsOrgAdmin -from orgs.mixins.api import OrgBulkModelViewSet +from applications.models import Application from perms.models import ApplicationPermission from perms import serializers +from ..base import BasePermissionViewSet -class ApplicationPermissionViewSet(OrgBulkModelViewSet): +class ApplicationPermissionViewSet(BasePermissionViewSet): """ 应用授权列表的增删改查API """ @@ -14,7 +14,6 @@ class ApplicationPermissionViewSet(OrgBulkModelViewSet): serializer_class = serializers.ApplicationPermissionSerializer filter_fields = ['name', 'category', 'type'] search_fields = filter_fields - permission_classes = (IsOrgAdmin,) def get_queryset(self): queryset = super().get_queryset().prefetch_related( @@ -22,3 +21,22 @@ class ApplicationPermissionViewSet(OrgBulkModelViewSet): ) return queryset + def filter_application(self, queryset): + application_id = self.request.query_params.get('application_id') + application_name = self.request.query_params.get('application') + if application_id: + applications = Application.objects.filter(pk=application_id) + elif application_name: + applications = Application.objects.filter(name=application_name) + else: + return queryset + if not applications: + return queryset.none() + queryset = queryset.filter(applications=applications) + return queryset + + def filter_queryset(self, queryset): + queryset = super().filter_queryset(queryset) + queryset = self.filter_application(queryset) + return queryset + diff --git a/apps/perms/api/asset/asset_permission.py b/apps/perms/api/asset/asset_permission.py index b57321a8a..67e2ce55d 100644 --- a/apps/perms/api/asset/asset_permission.py +++ b/apps/perms/api/asset/asset_permission.py @@ -2,14 +2,12 @@ # from django.db.models import Q -from common.permissions import IsOrgAdmin -from orgs.mixins.api import OrgBulkModelViewSet -from common.utils import get_object_or_none from perms.models import AssetPermission from perms.hands import ( - User, UserGroup, Asset, Node, SystemUser, + Asset, Node ) from perms import serializers +from ..base import BasePermissionViewSet __all__ = [ @@ -17,14 +15,13 @@ __all__ = [ ] -class AssetPermissionViewSet(OrgBulkModelViewSet): +class AssetPermissionViewSet(BasePermissionViewSet): """ 资产授权列表的增删改查api """ model = AssetPermission serializer_class = serializers.AssetPermissionSerializer filter_fields = ['name'] - permission_classes = (IsOrgAdmin,) def get_queryset(self): queryset = super().get_queryset().prefetch_related( @@ -32,35 +29,6 @@ class AssetPermissionViewSet(OrgBulkModelViewSet): ) return queryset - def is_query_all(self): - query_all = self.request.query_params.get('all', '1') == '1' - return query_all - - def filter_valid(self, queryset): - valid_query = self.request.query_params.get('is_valid', None) - if valid_query is None: - return queryset - invalid = valid_query in ['0', 'N', 'false', 'False'] - if invalid: - queryset = queryset.invalid() - else: - queryset = queryset.valid() - return queryset - - def filter_system_user(self, queryset): - system_user_id = self.request.query_params.get('system_user_id') - system_user_name = self.request.query_params.get('system_user') - if system_user_id: - system_user = get_object_or_none(SystemUser, pk=system_user_id) - elif system_user_name: - system_user = get_object_or_none(SystemUser, name=system_user_name) - else: - return queryset - if not system_user: - return queryset.none() - queryset = queryset.filter(system_users=system_user) - return queryset - def filter_node(self, queryset): node_id = self.request.query_params.get('node_id') node_name = self.request.query_params.get('node') @@ -112,55 +80,8 @@ class AssetPermissionViewSet(OrgBulkModelViewSet): ).distinct() return queryset - def filter_user(self, queryset): - user_id = self.request.query_params.get('user_id') - username = self.request.query_params.get('username') - if user_id: - user = get_object_or_none(User, pk=user_id) - elif username: - user = get_object_or_none(User, username=username) - else: - return queryset - if not user: - return queryset.none() - if not self.is_query_all(): - queryset = queryset.filter(users=user) - return queryset - groups = user.groups.all() - queryset = queryset.filter( - Q(users=user) | Q(user_groups__in=groups) - ).distinct() - return queryset - - def filter_user_group(self, queryset): - user_group_id = self.request.query_params.get('user_group_id') - user_group_name = self.request.query_params.get('user_group') - if user_group_id: - group = get_object_or_none(UserGroup, pk=user_group_id) - elif user_group_name: - group = get_object_or_none(UserGroup, name=user_group_name) - else: - return queryset - if not group: - return queryset.none() - queryset = queryset.filter(user_groups=group) - return queryset - - def filter_keyword(self, queryset): - keyword = self.request.query_params.get('search') - if not keyword: - return queryset - queryset = queryset.filter(name__icontains=keyword) - return queryset - def filter_queryset(self, queryset): queryset = super().filter_queryset(queryset) - queryset = self.filter_valid(queryset) - queryset = self.filter_user(queryset) - queryset = self.filter_keyword(queryset) queryset = self.filter_asset(queryset) queryset = self.filter_node(queryset) - queryset = self.filter_system_user(queryset) - queryset = self.filter_user_group(queryset) - queryset = queryset.distinct() return queryset diff --git a/apps/perms/api/base.py b/apps/perms/api/base.py index d4ffc9246..50cb96eae 100644 --- a/apps/perms/api/base.py +++ b/apps/perms/api/base.py @@ -1,13 +1,102 @@ from django.db.models import F -from orgs.mixins.api import OrgBulkModelViewSet from orgs.mixins.api import OrgRelationMixin +from django.db.models import Q +from common.permissions import IsOrgAdmin +from common.utils import get_object_or_none +from orgs.mixins.api import OrgBulkModelViewSet +from assets.models import SystemUser +from users.models import User, UserGroup __all__ = [ - 'RelationViewSet' + 'RelationViewSet', 'BasePermissionViewSet' ] +class BasePermissionViewSet(OrgBulkModelViewSet): + permission_classes = (IsOrgAdmin,) + + def filter_valid(self, queryset): + valid_query = self.request.query_params.get('is_valid', None) + if valid_query is None: + return queryset + invalid = valid_query in ['0', 'N', 'false', 'False'] + if invalid: + queryset = queryset.invalid() + else: + queryset = queryset.valid() + return queryset + + def is_query_all(self): + query_all = self.request.query_params.get('all', '1') == '1' + return query_all + + def filter_user(self, queryset): + user_id = self.request.query_params.get('user_id') + username = self.request.query_params.get('username') + if user_id: + user = get_object_or_none(User, pk=user_id) + elif username: + user = get_object_or_none(User, username=username) + else: + return queryset + if not user: + return queryset.none() + if not self.is_query_all(): + queryset = queryset.filter(users=user) + return queryset + groups = user.groups.all() + queryset = queryset.filter( + Q(users=user) | Q(user_groups__in=groups) + ).distinct() + return queryset + + def filter_keyword(self, queryset): + keyword = self.request.query_params.get('search') + if not keyword: + return queryset + queryset = queryset.filter(name__icontains=keyword) + return queryset + + def filter_system_user(self, queryset): + system_user_id = self.request.query_params.get('system_user_id') + system_user_name = self.request.query_params.get('system_user') + if system_user_id: + system_user = get_object_or_none(SystemUser, pk=system_user_id) + elif system_user_name: + system_user = get_object_or_none(SystemUser, name=system_user_name) + else: + return queryset + if not system_user: + return queryset.none() + queryset = queryset.filter(system_users=system_user) + return queryset + + def filter_user_group(self, queryset): + user_group_id = self.request.query_params.get('user_group_id') + user_group_name = self.request.query_params.get('user_group') + if user_group_id: + group = get_object_or_none(UserGroup, pk=user_group_id) + elif user_group_name: + group = get_object_or_none(UserGroup, name=user_group_name) + else: + return queryset + if not group: + return queryset.none() + queryset = queryset.filter(user_groups=group) + return queryset + + def filter_queryset(self, queryset): + queryset = super().filter_queryset(queryset) + queryset = self.filter_valid(queryset) + queryset = self.filter_user(queryset) + queryset = self.filter_system_user(queryset) + queryset = self.filter_user_group(queryset) + queryset = self.filter_keyword(queryset) + queryset = queryset.distinct() + return queryset + + class RelationViewSet(OrgRelationMixin, OrgBulkModelViewSet): def get_queryset(self): queryset = super().get_queryset()