django restframework使用redis实现token认证

目录

• 一、前言
• 二、详解
  • 1. 前期准备
  • 2. 配置redis
  • 3. 将token写入redis
    • 3.1 原来的登录代码
    • 3.2 重写后的登录代码
    • 3.3 登录后redis存储的用户记录
  • 4. 重写认证token方法
    • 4.1 源码分析
    • 4.2 进行重写
    • 4.3 加入认证配置
    • 4.4 效果展示
• 三、总结

pip install django-redis -i https://pypi.tuna.tsinghua.edu.cn/simple

# redis缓存配置
CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": "redis://127.0.0.1:6379/",
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
            "COMPRESSOR": "django_redis.compressors.zlib.ZlibCompressor",
            "IGNORE_EXCEPTIONS": True,
        }
    }
}

from django_redis import get_redis_connection
redis_connect = get_redis_connection()

@api_view(["POST"])
def login(request):
    """
    登录接口
    """
    user = authenticate(username=request.data["username"], password=request.data["password"])
    if user:
        Token.objects.filter(user_id=user.id).delete()
        token = Token.objects.create(user=user)
        _dict = {"id": user.id, "username": user.username, "first_name": user.first_name,
                 "last_name": user.last_name, "email": user.email}        
        redis_connect.set(token.key, json.dumps(_dict), 259200)  # 存redis 259200秒=72个小时
        return Response(data={"status_code": 200, "msg": "登录成功！", "token": token.key})
    return Response(data={"status_code": 403, "msg": "密码错误！"})

@api_view(["POST"])
def login(request):
    """
    登录接口
    """
    user = authenticate(username=request.data["username"], password=request.data["password"])
    if user:
        token = binascii.hexlify(os.urandom(20)).decode()  # 生成token 的方式
        _dict = {"id": user.id, "username": user.username, "first_name": user.first_name,
                 "last_name": user.last_name, "email": user.email}
        redis_connect.set(token, json.dumps(_dict), 259200)  # 存redis 259200秒=72个小时
        return Response(data={"status_code": 200, "msg": "登录成功！", "token": token})
    return Response(data={"status_code": 403, "msg": "密码错误！"})

    def authenticate_credentials(self, key):
        model = self.get_model()
        try:
            token = model.objects.select_related("user").get(key=key)
        except model.DoesNotExist:
            raise exceptions.AuthenticationFailed(_("Invalid token."))

        if not token.user.is_active:
            raise exceptions.AuthenticationFailed(_("User inactive or deleted."))

        return (token.user, token)

        try:
            token = model.objects.select_related("user").get(key=key)
        except model.DoesNotExist:
            raise exceptions.AuthenticationFailed(_("Invalid token."))

      if not token.user.is_active:
            raise exceptions.AuthenticationFailed(_("User inactive or deleted."))

class RedisTokenAuthentication(TokenAuthentication):

    def authenticate_credentials(self, key):
        null = None  # json的None为null，所以需要定义一下
        user_data = redis_connect.get(key)
        if user_data:
            user_dict = json.loads(user_data)
            user_obj = User()
            for key_name in user_dict.keys():
                setattr(user_obj, key_name, user_dict[key_name])
            return user_obj, key
        raise exceptions.AuthenticationFailed(_("无效的token."))

REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": (  #如果有REST_FRAMEWORK配置项了单独加入该项即可
        "Demo.RedisAuthentication.RedisTokenAuthentication",  # 项目名称.重新认证类所在的文件.类名
    ),
}

path("test-token", views.test_token),

@api_view(["GET"])
@permission_classes((permissions.IsAuthenticated,))
def test_token(request):
    """
    测试token
    """
    print("登录的用户名是:", request.user)
    res_data = {"data": {"status_code": 200}}
    return Response(**res_data)