如何使用 Django 中具有不同权限的嵌套模型对嵌套序列化程序进行自定义权限检查

How to do custom permission check for nested serializer with nested models having different permission in django

我正在构建一个网络应用程序,用户可以在其中创建团队并添加成员和 his/her 项目。一切正常,但现在是许可部分。一个模型是 Team,另一个是 Project。现在我已经为扩展 BasePermission.

的两个模型编写了自定义权限

operation/permission 将是:

  1. User1创建了一个团队Team1,可以添加任何成员并添加他的项目(没有添加其他项目的权限)

  2. Team1 的成员可以添加自己的项目并编辑 (CRU) 其他人添加的项目。成员没有删除Team1的权限,只有创建者可以删除团队。

  3. 项目只能由其所在团队的成员编辑。其他人不能。只有项目的创建者可以删除它。

权限:

from rest_framework import permissions
from .models import Team,Project
from rest_framework import serializers

class ProjectPermission(permissions.BasePermission):
    message = "You do not have permission to perform this action with Project that doesn't belong to you or you are not a member of the team for this Project"
    
    def has_object_permission(self, request,view, obj):
        if not request.method in permissions.SAFE_METHODS:
            if request.method != "DELETE":
                if obj.team: #Team can be null when creating a project
                    return obj.created_by == request.user or request.user in obj.team.members.all() 
                return obj.created_by == request.user
            return obj.created_by == request.user
        return request.user.is_authenticated
        
    def has_permission(self, request, view):
        return request.user.is_authenticated

class TeamPermission(permissions.BasePermission):
    message = "You do not have permission to perform this action with Team that is not created by you or you are not a member with full permission"
    
    def has_object_permission(self, request, view, obj):
        if request.method in permissions.SAFE_METHODS:
            return request.user.is_authenticated
        else:
            if request.method != "DELETE":
                if obj.created_by == request.user or request.user in obj.members.all():
                    return True
            return obj.created_by == request.user
    
    def has_permission(self, request, view):
        if not request.method in permissions.SAFE_METHODS:
            
            if request.method =="DELETE":
                return True
            Projects = request.data.get('Projects')
            if request.method =="PUT" or request.method == "PATCH":
                if len(Projects)==0:return True # for removing a list projects an empty array is passed, in the serializer set function "remove set(current projects - empty list)" is used to identify the projects to remove 
            if Projects:
                perm = []
                for Project in Projects:
                    try:
                         #this is the issue, if there are 100 projects to be added, it does 100 queries to fetch the object
                        perm.append(ProjectPermission().has_object_permission(request, view,Project.objects.get(id=Project)))
                    except Exception as e:
                        raise serializers.ValidationError(e)
                if False in perm:
                    raise serializers.ValidationError(
                        {"detail":"You do not have permission to perform this action with Project that doesn't belong to you or you are not a member of the team for this Project"})           
                return True
        else:
            return request.user.is_authenticated

最后这一行导致 100 个查询到 100 个项目的数据库。我可以按项目列表进行过滤。但是还有其他选择吗?

perm.append(ProjectPermission().has_object_permission(request, view,Project.objects.get(id=Project))

观看次数:

class ProjectView(viewsets.ModelViewSet):
    '''
    Returns a list of all the Projects. created by user and others
    Supports CRUD
    '''
    queryset=Project.objects.select_related('team','created_by').all()
    serializer_class=ProjectSerializer
    permission_classes=[ProjectPermission]

class TeamView(viewsets.ModelViewSet):
    """
    Returns a list of all the teams. created by user and others
    Supports CRUD
    """
    queryset=Team.objects.prefetch_related('members','projects').select_related('created_by').all()
    serializer_class=TeamSerializer
    permission_classes=[TeamPermission]

型号:

class Team(models.Model):
    team_name=models.CharField(max_length=50,blank=False,null=False,unique=True)
    created_by=models.ForeignKey(User,on_delete=models.SET_NULL,null=True)
    created_at=models.DateTimeField(auto_now_add=True)
    members=models.ManyToManyField(User,related_name='members')

    def __str__(self) -> str:
        return self.team_name
    
    class Meta:
        ordering=['-created_at']
     
class Project(models.Model):
    team=models.ForeignKey(Team,related_name='projects',blank=True,null=True,on_delete=models.SET_NULL)
    project_name=models.CharField(max_length=50,blank=False,null=False,unique=True)
    description=models.CharField(max_length=1000,null=True,blank=True)
    created_by=models.ForeignKey(User,on_delete=models.SET_NULL,null=True)
    created_at=models.DateTimeField(auto_now_add=True)
    file_name=models.CharField(max_length=100,null=True,blank=True)

    def __str__(self) -> str:
        return self.project_name
    
    class Meta:
        ordering = ['-created_at']

有没有一种有效的方法来实现我所需要的?我阅读了有关在模型中添加权限的信息,但我不知道在这种情况下如何完成。任何建议都会有很大的帮助

我们分解一下,任何User都可以创建一个Team,所以有一个权限IsAuthenticated

任何 Team 成员都可以 CRU 一个项目,所以这里有三个权限:

IsAuthenticated & ( IsTeamOwner | IsTeamMember)

删除项目有两个权限(IsAuthenticated & IsProjectOwner)等等

所以,权限可能是这样的:

class IsTeamOwner(BasePermission):
    message = "You do not have permission to perform this action"

    def has_object_permission(self, request, view, obj):
        return obj.created_by == request.user


class IsProjectOwner(BasePermission):
    message = "You do not have permission to perform this action"

    def has_object_permission(self, request, view, obj):
        return obj.created_by == request.user


class IsTeamMember(BasePermission):
    message = "You do not have permission to perform this action"

    def has_object_permission(self, request, view, obj):
        return request.user in obj.members.all()

更新

检查项目id是否存在,是否由当前用户创建:

try: 
    project = Project.objects.only('team').get(id=recieved_id, created_by=request.user)
except Project.DoesNotExist:
    raise ProjectNotExist() # Create this exception

检查项目id是否存在以及是否属于当前用户所在的团队:

try: 
    project = Project.objects.get(id=recieved_id).select_related('team')# try to use prefetch_related('team__members') also, I don't know it would work or not 
    if request.user not in project.team.members.all():
        raise ProjectNotExist() 
except Project.DoesNotExist:
    raise ProjectNotExist()

为了进行更多自定义,在检查项目是否存在后,您可以使用 bulk_update 仅通过一个查询来更新每个项目中的 team 字段。