Django多级用户角色权限管理实现咨询：客户数据隔离

一、模型补全与修正 #

根据需求，现有模型需要补充CustomerCustomUser关联表，同时给CustomUser添加Customer外键，确保用户与客户的关联关系：

from django.contrib.auth.models import Permission, Group
from django.contrib.auth.base_user import BaseUserManager, AbstractBaseUser
from django.contrib.auth.models import PermissionsMixin
from django.db import models
from django.core.validators import RegexValidator
from django.core.exceptions import ValidationError

mobile_regex = RegexValidator(regex=r'^\+?1?\d{9,15}$', message="手机号码格式错误，需为'+999999999'形式，最多15位数字")

class Role(models.Model):
    name = models.CharField(max_length=100, unique=True)
    permissions = models.ManyToManyField(Permission)

    def __str__(self):
        return self.name
    
class CustomUserManager(BaseUserManager):
    def create_user(self, username, email, password=None):
        if not email:
            raise ValueError("必须填写邮箱字段")
        email = self.normalize_email(email)
        user = self.model(username=username, email=email)
        user.set_password(password)
        user.save(using=self._db)
        return user
    
    def create_superuser(self, username, email, password=None):
        user = self.create_user(username, email, password)
        user.is_staff = True
        user.is_superuser = True
        user.save(using=self._db)
        return user

class CustomUser(AbstractBaseUser, PermissionsMixin):
    username = models.CharField(max_length=150, unique=True)
    email = models.EmailField(unique=True)
    is_active = models.BooleanField(default=True)
    is_staff = models.BooleanField(default=False)
    role = models.ForeignKey(Role, on_delete=models.CASCADE, related_name='custom_users')
    customer = models.ForeignKey('Customer', on_delete=models.SET_NULL, null=True, blank=True, related_name='assigned_users')
    groups = models.ManyToManyField(Group, blank=True, related_name='custom_users')
    user_permissions = models.ManyToManyField(
        Permission,
        blank=True,
        related_name='custom_users_permissions'
    )

    objects = CustomUserManager()
    USERNAME_FIELD = 'email'
    REQUIRED_FIELDS = ['username']

    def __str__(self):
        return self.username

    def clean(self):
        super().clean()
        # 模型层验证：非超级管理员必须关联客户
        if not self.is_superuser and not self.customer:
            raise ValidationError("普通用户必须关联客户")

class AbstractPerson(models.Model):
    first_name = models.CharField(max_length=60)
    last_name = models.CharField(max_length=60)
    email = models.EmailField(unique=True)
    mobile = models.CharField(max_length=10, validators=[mobile_regex])
    muser_id = models.IntegerField(null=True, blank=True)

    class Meta:
        abstract = True

class Customer(AbstractPerson):
    custom_users = models.ManyToManyField(CustomUser, through='CustomerCustomUser')

# 补充关联表：关联客户、用户与角色
class CustomerCustomUser(models.Model):
    customer = models.ForeignKey(Customer, on_delete=models.CASCADE)
    custom_user = models.ForeignKey(CustomUser, on_delete=models.CASCADE)
    role = models.ForeignKey(Role, on_delete=models.CASCADE)

    class Meta:
        unique_together = ('customer', 'custom_user', 'role')  # 避免重复关联

二、Admin核心配置（复用默认权限+数据隔离） #

封装基础Admin类实现数据过滤与权限校验，其他模型继承该类即可：

from django.contrib import admin
from django.contrib.auth.admin import UserAdmin as BaseUserAdmin
from .models import Customer, CustomUser, Role, CustomerCustomUser

class CustomerRestrictedAdmin(admin.ModelAdmin):
    # 过滤查询集：仅展示当前用户关联客户的记录
    def get_queryset(self, request):
        qs = super().get_queryset(request)
        if request.user.is_superuser:
            return qs
        # 仅对关联了Customer的用户生效
        if hasattr(request.user, 'customer') and request.user.customer:
            if self.model == Customer:
                return qs.filter(id=request.user.customer.id)
            elif self.model == CustomUser:
                return qs.filter(customer=request.user.customer)
            elif self.model == CustomerCustomUser:
                return qs.filter(customer=request.user.customer)
        return qs.none()

    # 复用Django默认的增/改/删/查权限
    def has_add_permission(self, request):
        if request.user.is_superuser:
            return True
        return request.user.has_perm(f'{self.model._meta.app_label}.add_{self.model._meta.model_name}')

    def has_change_permission(self, request, obj=None):
        if request.user.is_superuser:
            return True
        if obj and hasattr(request.user, 'customer') and obj.customer == request.user.customer:
            return request.user.has_perm(f'{self.model._meta.app_label}.change_{self.model._meta.model_name}')
        return False

    def has_delete_permission(self, request, obj=None):
        if request.user.is_superuser:
            return True
        if obj and hasattr(request.user, 'customer') and obj.customer == request.user.customer:
            return request.user.has_perm(f'{self.model._meta.app_label}.delete_{self.model._meta.model_name}')
        return False

    def has_view_permission(self, request, obj=None):
        if request.user.is_superuser:
            return True
        if obj:
            return hasattr(request.user, 'customer') and obj.customer == request.user.customer and request.user.has_perm(f'{self.model._meta.app_label}.view_{self.model._meta.model_name}')
        return request.user.has_perm(f'{self.model._meta.app_label}.view_{self.model._meta.model_name}')

# CustomUser Admin：继承Django默认UserAdmin+权限隔离类
class CustomUserAdmin(BaseUserAdmin, CustomerRestrictedAdmin):
    list_display = ('username', 'email', 'customer', 'role', 'is_active', 'is_staff')
    list_filter = ('customer', 'role', 'is_active', 'is_staff')
    fieldsets = (
        (None, {'fields': ('username', 'email', 'password')}),
        ('权限配置', {'fields': ('is_active', 'is_staff', 'is_superuser', 'role', 'groups', 'user_permissions')}),
        ('客户关联', {'fields': ('customer',)}),
    )
    add_fieldsets = (
        (None, {
            'classes': ('wide',),
            'fields': ('username', 'email', 'password1', 'password2', 'customer', 'role', 'is_active', 'is_staff')}
        ),
    )

    # 限制普通用户只能选择自己关联的客户
    def formfield_for_foreignkey(self, db_field, request, **kwargs):
        if db_field.name == 'customer' and not request.user.is_superuser:
            if hasattr(request.user, 'customer'):
                kwargs['queryset'] = Customer.objects.filter(id=request.user.customer.id)
        # 可选：限制普通用户只能分配特定角色
        if db_field.name == 'role' and not request.user.is_superuser:
            # 此处可根据业务逻辑过滤角色，比如只允许分配客户拥有的角色
            kwargs['queryset'] = Role.objects.all()
        return super().formfield_for_foreignkey(db_field, request, **kwargs)

    # 强制新增用户关联当前客户（非超级管理员）
    def save_model(self, request, obj, form, change):
        if not request.user.is_superuser and not change:
            obj.customer = request.user.customer
        super().save_model(request, obj, form, change)

# Customer Admin
class CustomerAdmin(CustomerRestrictedAdmin):
    list_display = ('first_name', 'last_name', 'email', 'mobile')

    # 普通用户只能修改自身客户的部分字段
    def get_readonly_fields(self, request, obj=None):
        if not request.user.is_superuser and obj and obj.id == request.user.customer.id:
            return ('email',)  # 禁止修改邮箱（避免与用户关联冲突）
        return super().get_readonly_fields(request, obj)

# Role Admin：管理角色与权限关联
class RoleAdmin(CustomerRestrictedAdmin):
    list_display = ('name',)
    filter_horizontal = ('permissions',)  # 可视化多选权限

# CustomerCustomUser Admin：管理客户-用户-角色关联
class CustomerCustomUserAdmin(CustomerRestrictedAdmin):
    list_display = ('customer', 'custom_user', 'role')
    list_filter = ('customer', 'role')

# 注册Admin
admin.site.register(Customer, CustomerAdmin)
admin.site.register(CustomUser, CustomUserAdmin)
admin.site.register(Role, RoleAdmin)
admin.site.register(CustomerCustomUser, CustomerCustomUserAdmin)

三、角色与Django默认权限对齐 #

通过信号同步角色权限到用户，确保角色权限与Django默认权限系统一致：

from django.db.models.signals import post_save
from django.dispatch import receiver

@receiver(post_save, sender=CustomUser)
def sync_role_permissions(sender, instance, created, **kwargs):
    if instance.role:
        # 同步角色权限到用户的权限列表
        instance.user_permissions.set(instance.role.permissions.all())
    # 避免递归保存
    post_save.disconnect(sync_role_permissions, sender=sender)
    instance.save()
    post_save.connect(sync_role_permissions, sender=sender)

四、测试验证 #

1. 超级管理员：可查看、操作所有模型的所有记录
2. 客户关联用户：登录后仅能查看自身关联客户的所有记录，仅能创建/修改/删除该客户下的用户，操作权限受角色关联的Django默认权限控制
3. 权限校验：给角色分配add_customuser权限后，拥有该角色的用户即可添加新用户，无权限则无法操作

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