Django rest_framework用户认证和权限

完整的代码

　　https://gitee.com/mom925/django-system

使用jwt实现用户认证

    pip install djangorestframework-simplejwt

重新定义一下User类

class Users(AbstractUser):
        class Meta:
            db_table = "system_users"
            verbose_name = "用户表"
            verbose_name_plural = verbose_name
            ordering = ("-create_datetime",)
        GENDER_CHOICES = (
            (0, "未知"),
            (1, "男"),
            (2, "女"),
        )
        username = models.CharField(max_length=150, unique=True, db_index=True, verbose_name="用户账号", help_text="用户账号")
        nickname = models.CharField(max_length=50, verbose_name="昵称", null=True, blank=True, help_text="昵称")
        email = models.EmailField(max_length=255, verbose_name="邮箱", null=True, blank=True, help_text="邮箱")
        mobile = models.CharField(max_length=255, verbose_name="电话", null=True, blank=True, help_text="电话")
        avatar = models.CharField(max_length=255, verbose_name="头像", null=True, blank=True, help_text="头像")
        gender = models.IntegerField(choices=GENDER_CHOICES, default=0, verbose_name="性别", null=True, blank=True, help_text="性别")
        create_datetime = models.DateTimeField(auto_now_add=True, verbose_name="创建时间", null=True, blank=True, help_text="创建时间")
        role = models.ManyToManyField(to="Role", blank=True, verbose_name="关联角色", db_constraint=False, help_text="关联角色")

并在 settings.py 加入

    AUTH_USER_MODEL = 'system.Users'
    USERNAME_FIELD = "username"

设置全局的默认认证和权限 在settings.py 配置 REST_FRAMEWORK

REST_FRAMEWORK = {
        "DEFAULT_PERMISSION_CLASSES": [
                "rest_framework.permissions.IsAuthenticated",  # 只有经过身份认证确定用户身份才能访问
            ],
        'DEFAULT_AUTHENTICATION_CLASSES': [
                "rest_framework_simplejwt.authentication.JWTAuthentication",
        ],
    }

接下来需要获取用户的token

先配置获取token和刷新token的路由

path('login/', views.LoginView.as_view(), name='登录'),
path('refresh/', views.CustomTokenRefreshView.as_view(), name='刷新token'),

接下来重写一下登录的序列器

class LoginSerializer(TokenObtainPairSerializer):
    """
    登录的序列化器:
    """

        class Meta:
            model = Users
            fields = "__all__"
            read_only_fields = ["id"]

        def validate(self, attrs):
            if settings.AUTH_CAPTCHA:
                captcha = self.initial_data.get("captcha", None)
                captchaKey = self.initial_data.get("captchaKey", None)
                if captcha is None:
                    raise CustomValidationError(detail='验证码不能为空')
                cache_captcha = cache.get('captcha_' + captchaKey)
                if cache_captcha is None:
                    raise CustomValidationError(detail='验证码过期')
                if str(captcha) != str(cache_captcha):
                    raise CustomValidationError(detail='验证码不正确')
                cache.delete('captcha_' + captchaKey)
            data = super().validate(attrs)
            data["user_name"] = self.user.username
            data["user_id"] = self.user.id
            data['avatar_url'] = get_avatar_url(self.user.avatar) if self.user.avatar else self.user.avatar
            request = self.context.get("request")
            request.user = self.user
            request_save_log(request, {'action': '用户登录', 'describe': '用户账号密码登录'})
            return {"code": 1, "msg": "success", "data": data}

登录视图

class LoginView(TokenObtainPairView):
    """
    tags:
        登录
    """
        serializer_class = LoginSerializer
        authentication_classes = []
        permission_classes = []

刷新token视图

class CustomTokenRefreshView(TokenRefreshView):
    """
    tags:
        刷新token
    """
        def post(self, request, *args, **kwargs):
            refresh_token = request.data.get("refresh")
            try:
                token = RefreshToken(refresh_token)
                data = {
                    "access": str(token.access_token),
                    "refresh": str(token)
                }
            except:
                return ErrorResponse(msg='error')
            return SuccessResponse(data=data)

自定义一下后台的认证，使用自定义后台认证需要在settings.py 中加入 AUTHENTICATION_BACKENDS = ["utils.backend.CustomBackend"]

class CustomBackend(ModelBackend):
    """
    Django原生认证方式
    """

        def authenticate(self, request, username=None, password=None, **kwargs):

            if username is None:
                username = kwargs.get(UserModel.USERNAME_FIELD)
            if username is None or password is None:
                return
            try:
                user = UserModel._default_manager.get_by_natural_key(username)
            except Exception:
                pass
                # print(traceback.format_exc())
            # except UserModel.DoesNotExist:
            #     # Run the default password hasher once to reduce the timing
            #     # difference between an existing and a nonexistent user (#20760).
            #     UserModel().set_password(password)
            else:
                if user.check_password(password):
                    if self.user_can_authenticate(user):
                        return user
                    else:
                        raise CustomValidationError("当前用户已被禁用，请联系客服或管理员!")

登录成功

自定义权限

    定义权限和角色模型

class Role(models.Model):
    class Meta:
        db_table = "system_role"
        verbose_name = "角色表"
        verbose_name_plural = verbose_name
        ordering = ("sort",)
    name = models.CharField(max_length=64, verbose_name="角色名称", help_text="角色名称")
    sort = models.IntegerField(default=1, verbose_name="角色顺序", help_text="角色顺序")
    status = models.BooleanField(default=True, verbose_name="角色状态", help_text="角色状态")
    admin = models.BooleanField(default=False, verbose_name="是否为admin", help_text="是否为admin")
    remark = models.TextField(verbose_name="备注", help_text="备注", null=True, blank=True)
    permission = models.ManyToManyField(to="Permission", verbose_name="关联权限", db_constraint=False, help_text="关联权限")


class Permission(models.Model):
    class Meta:
        db_table = "system_permission"
        verbose_name = "权限表"
        verbose_name_plural = verbose_name
        ordering = ("-value",)

    METHOD_CHOICES = (
        (0, "GET"),
        (1, "POST"),
        (2, "PUT"),
        (3, "DELETE"),
    )
    name = models.CharField(max_length=64, verbose_name="名称", help_text="名称")
    value = models.CharField(max_length=64, verbose_name="权限值", help_text="权限值")
    api = models.CharField(max_length=200, verbose_name="接口地址", help_text="接口地址")
    method = models.IntegerField(default=0, verbose_name="接口请求方法", null=True, blank=True, help_text="接口请求方法")


class ApiWhiteList(models.Model):
    class Meta:
        db_table = "api_white_list"
        verbose_name = "接口白名单"
        verbose_name_plural = verbose_name
        ordering = ("-create_datetime",)

    METHOD_CHOICES = (
        (0, "GET"),
        (1, "POST"),
        (2, "PUT"),
        (3, "DELETE"),
    )
    url = models.CharField(max_length=200, help_text="url地址", verbose_name="url")

    method = models.IntegerField(default=0, verbose_name="接口请求方法", null=True, blank=True, help_text="接口请求方法")
    enable_datasource = models.BooleanField(default=True, verbose_name="激活数据权限", help_text="激活数据权限", blank=True)
    create_datetime = models.DateTimeField(auto_now_add=True, verbose_name="创建时间", null=True, blank=True, help_text="创建时间")

自定义的权限可以继承 BasePermission 自己定义其中的逻辑，例如：

　　

class CustomPermission(BasePermission):
            """自定义权限"""

            def has_permission(self, request, view):
                if isinstance(request.user, AnonymousUser):
                    return False
                # 判断是否是超级管理员
                if request.user.is_superuser:
                    return True
                else:
                    api = request.path  # 当前请求接口
                    method = request.method  # 当前请求方法
                    methodList = ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS', 'PATCH']
                    method = methodList.index(method)
                    # ***接口白名单***
                    api_white_list = ApiWhiteList.objects.filter(enable_datasource=True).values(permission__api=F('url'), permission__method=F('method'))
                    api_white_list = [
                        str(item.get('permission__api').replace('{id}', '([a-zA-Z0-9-]+)')) + ":" + str(
                            methodList.index(item.get('permission__method', 'GET'))) + '$' for item in api_white_list if item.get('permission__api')]
                    # ********#
                    if not hasattr(request.user, "role"):
                        return False
                    userApiList = request.user.role.values('permission__api', 'permission__method')  # 获取当前用户的角色拥有的所有接口
                    ApiList = [
                        str(item.get('permission__api').replace('{id}', '([a-zA-Z0-9-]+)')) + ":" + str(
                            item.get('permission__method')) + '$' for item in userApiList if item.get('permission__api')]
                    new_api_ist = api_white_list + ApiList
                    new_api = api + ":" + str(method)
                    # print(api, new_api, method, new_api_ist)
                    for item in new_api_ist:
                        matchObj = re.match(item, new_api, re.M | re.I)
                        if matchObj is None:
                            continue
                        else:
                            return True
                    else:
                        return False

使用时可以在视图的 permission_classes = [CustomPermission] 加入到列表中