Django Rest Framework自定义权限校验：验证用户是否归属对应组织

1. 调整自定义权限类 #

你需要同时实现has_permission（请求进入视图前的全局校验，覆盖列表查询、创建等无单对象的操作）和has_object_permission（单对象读写更新的校验），逻辑如下：

from rest_framework import permissions
from .models import UserOrganization

class OrgAccessPermission(permissions.BasePermission):
    def has_permission(self, request, view):
        # 先校验登录状态
        if not request.user.is_authenticated:
            return False
        # 管理员直接放行
        if request.user.is_admin:
            return True
        # 从请求中获取携带的组织ID，可根据你的实际传参方式调整取值逻辑
        org_id = self._get_org_id(request)
        if not org_id:
            return False
        # 校验用户是否为该组织成员
        return UserOrganization.objects.filter(
            user=request.user,
            organization_id=org_id
        ).exists()

    def has_object_permission(self, request, view, obj):
        if request.user.is_admin:
            return True
        org_id = self._get_org_id(request)
        # 校验操作的集成所属组织和当前请求的组织一致
        if str(obj.organization_id) != str(org_id):
            return False
        # 保留原有的用户归属校验逻辑
        if view.action in ['retrieve', 'update', 'partial_update']:
            return obj.user == request.user
        elif view.action == 'destroy':
            return False
        return False

    def _get_org_id(self, request):
        # 可根据实际传参方式调整：请求头、请求体、查询参数的优先级
        return request.META.get('HTTP_X_ORGANIZATION_ID') \
            or request.data.get('organization_id') \
            or request.query_params.get('org_id')

注意：Django 3.0+ 版本中is_authenticated是属性无需加括号，如果你使用的是旧版本可恢复为request.user.is_authenticated()

2. 调整IntegrationViewSet配置 #

需要补充组织维度的查询过滤，以及创建时关联对应组织：

class IntegrationViewSet(mixins.CreateModelMixin,
                         mixins.ListModelMixin,
                         mixins.RetrieveModelMixin,
                         mixins.UpdateModelMixin,
                         viewsets.GenericViewSet):
    # 绑定自定义权限类
    permission_classes = [OrgAccessPermission]
    serializer_class = IntegrationSerializer

    def get_queryset(self):
        user = self.request.user
        org_id = self._get_org_id()
        if user.is_admin:
            return Integration.objects.filter(organization_id=org_id)
        # 普通用户仅能查询自己在当前组织下创建的集成
        return Integration.objects.filter(user=user, organization_id=org_id)

    def perform_create(self, serializer):
        org_id = self._get_org_id()
        # 保存集成时同时关联用户和所属组织
        serializer.save(
            user=self.request.user,
            organization_id=org_id
        )
    
    def _get_org_id(self):
        # 和权限类的取值逻辑保持一致
        return self.request.META.get('HTTP_X_ORGANIZATION_ID') \
            or self.request.data.get('organization_id') \
            or self.request.query_params.get('org_id')

额外说明 #

• 前提是你的Integration模型已经定义了organization外键字段，关联到Organization模型
• 如果需要区分组织内角色权限（比如仅Owner可删除集成），可以在校验成员时增加role字段的过滤条件
• 确保前端传参的组织ID位置和代码中的取值逻辑匹配，避免取不到ID导致校验失败

内容的提问来源于stack exchange，提问作者Hayden