Django多角色权限控制与QuerySet自动过滤方案咨询

Hey Kenneth, 针对你在Django邮件扫描应用中遇到的多角色权限过滤QuerySet问题，我整理了几个在实际项目中验证过的实用方案，帮你搞定这个需求：

你之前觉得在ModelManager里拿request.user不合理是完全正确的——ModelManager是全局层面的组件，和请求上下文耦合会导致代码不可复用（比如后台任务调用时就没有request），正确的过滤逻辑应该放在视图层/权限层，这里能安全获取当前用户。

方案1：自定义视图Mixin（无额外依赖，快速实现） #

这是最轻量化的方案，通过编写可复用的Mixin类，在视图中自动根据用户角色过滤QuerySet，适合你的三种角色场景：

第一步：编写Domain过滤Mixin #

from django.db.models import Q
from rest_framework import generics

class DomainFilterMixin:
    def get_queryset(self):
        user = self.request.user
        # 超级管理员直接返回全量数据
        if user.is_superuser:
            return super().get_queryset()
        
        # 获取用户关联的所有域
        user_domains = user.domains.all()
        # 无关联域则返回空集合
        if not user_domains.exists():
            return self.queryset.none()
        
        # 根据不同模型的字段做适配过滤
        model = self.queryset.model
        if hasattr(model, "domain"):
            # 比如黑白名单策略模型，直接关联单个domain
            return self.queryset.filter(domain__in=user_domains)
        elif hasattr(model, "sender_domain") and hasattr(model, "recipient_domain"):
            # 消息日志模型，过滤发件/收件域属于用户管理范围的记录
            return self.queryset.filter(
                Q(sender_domain__in=user_domains) | Q(recipient_domain__in=user_domains)
            )
        
        # 默认返回空（可根据新增模型扩展逻辑）
        return self.queryset.none()

第二步：在视图中复用Mixin #

class MessageLogListView(DomainFilterMixin, generics.ListAPIView):
    queryset = MessageLog.objects.all()
    serializer_class = MessageLogSerializer
    permission_classes = [IsAuthenticated]  # 确保用户已认证

class BlacklistPolicyView(DomainFilterMixin, generics.ListAPIView):
    queryset = BlacklistPolicy.objects.all()
    serializer_class = BlacklistPolicySerializer
    permission_classes = [IsAuthenticated]

方案2：使用django-rest-framework-access-policy（声明式配置，适合复杂场景） #

如果你的权限规则后续会更复杂，推荐用这个DRF扩展包，它支持声明式的权限规则+QuerySet过滤，代码可读性和可维护性更高：

第一步：安装并配置 #

pip install django-rest-framework-access-policy

第二步：定义访问策略类 #

from rest_access_policy import AccessPolicy
from django.db.models import Q

class DomainBasedAccessPolicy(AccessPolicy):
    # 声明权限规则
    statements = [
        {
            "action": ["list", "retrieve"],
            "principal": "superuser",
            "effect": "allow"
        },
        {
            "action": ["list", "retrieve"],
            "principal": "authenticated",
            "effect": "allow",
            "condition": "has_associated_domains"
        }
    ]

    # 自定义条件：检查用户是否关联了域
    def has_associated_domains(self, request, view, action):
        return request.user.domains.exists()

    # 自动过滤QuerySet
    def get_queryset(self, request, view):
        user = request.user
        if user.is_superuser:
            return view.queryset.all()
        
        user_domains = user.domains.all()
        model = view.queryset.model
        if hasattr(model, "domain"):
            return view.queryset.filter(domain__in=user_domains)
        elif hasattr(model, "sender_domain") and hasattr(model, "recipient_domain"):
            return view.queryset.filter(
                Q(sender_domain__in=user_domains) | Q(recipient_domain__in=user_domains)
            )
        return view.queryset.none()

第三步：在视图中使用 #

class MessageLogListView(generics.ListAPIView):
    queryset = MessageLog.objects.all()
    serializer_class = MessageLogSerializer
    permission_classes = [DomainBasedAccessPolicy]

方案3：结合django-guardian（细粒度对象权限场景） #

如果你后续需要给用户分配单个域的独立权限（而不是批量Many2Many关联），可以用django-guardian配合视图过滤：

核心代码示例 #

from guardian.shortcuts import get_objects_for_user

class MessageLogListView(generics.ListAPIView):
    queryset = MessageLog.objects.all()
    serializer_class = MessageLogSerializer
    permission_classes = [IsAuthenticated]

    def get_queryset(self):
        user = self.request.user
        if user.is_superuser:
            return super().get_queryset()
        
        # 获取用户有权限管理的所有域（需提前给用户分配view_domain权限）
        allowed_domains = get_objects_for_user(user, "your_app.view_domain")
        if not allowed_domains.exists():
            return self.queryset.none()
        
        return self.queryset.filter(
            Q(sender_domain__in=allowed_domains) | Q(recipient_domain__in=allowed_domains)
        )

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