django restframework 基于token实现用户认证详解

1. 本节主要讲述 django restframework 用户认证部分的内容

0. 环境配置

1. 基于token认证

2. 基于用户请求头认证

3. JWT认证

2. 环境配置

pip install django==2.0.5 pip install djangorestframework pip install pymysql pip install python-dateutil

2. 基于token认证

2.1 数据库模型设计：models.py

from django.db import models from datetime import datetime # Create your models here.

# 用户模型
class UserInfo(models.Model): user_type_choics = ( (1, "普通用户"), (2, "普通会员"), (3, "白金会员"), (4, "黄金会员"), ) username = models.CharField(max_length=64, verbose_name="用户名") password = models.CharField(max_length=64, verbose_name="密码") user_type = models.IntegerField(choices=user_type_choics) add_time = models.DateTimeField(default=datetime.now, verbose_name="注册时间") class Mate: managed = True db_table = "user_info" verbose_name = "用户模型" verbose_name_plural = verbose_name def __str__(self): return self.username # 存放用户登录成功后的token
class UserToken(models.Model): user = models.OneToOneField(to='UserInfo', unique=True, on_delete=False) token = models.CharField(max_length=64, verbose_name="用户token") expiration_time = models.DateTimeField(default=datetime.now, verbose_name="过期时间") add_time = models.DateTimeField(default=datetime.now, verbose_name="添加时间") class Mate: managed = True db_table = "user_token" verbose_name = "用户Token" verbose_name_plural = verbose_name def __str__(self): return self.token

2.2 同步数据库：数据同步好了之后添加几条用户信息

https://www.cnblogs.com/xingxingnbsp/p/10655255.html

2.3 路由：urls.py

from django.conf.urls import url from . import views urlpatterns = [ # 用户登录接口
    url(r'login/$',views.LoginView.as_view()), # 商品列表接口
    url(r'goods/$',views.GoodsView.as_view()), # 用户信息接口
    url(r'userinfo/$',views.UserInfoView.as_view()), ]

2.4 视图：views.py

import hashlib from datetime import datetime import dateutil.relativedelta import time from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import exceptions from rest_framework.authentication import BaseAuthentication from . import models from token_authentication.settings import SECRET_KEY # 基于token的用户认证
class TokenAuthtication(BaseAuthentication): def authenticate(self, request): # 1. 在请求头的query_params中获取token
        # token = request.query_params.get('token')

        # 2. 直接在请求头中获取token
        token = request._request.GET.get('token') token_obj = models.UserToken.objects.filter(token=token).first() if not token_obj: raise exceptions.AuthenticationFailed("用户认证失败") else: datetime_now = datetime.now() if token_obj.expiration_time > datetime_now: # 在 rest framework 内部会将两个字段赋值给request，以供后续操作使用
                return (token_obj.user, token_obj) else: raise exceptions.AuthenticationFailed("用户token过期,请重新登录") def authenticate_header(self, request): # 验证失败时，返回的响应头WWW-Authenticate对应的值
        pass


# 生成token
def md5(username): m = hashlib.md5(bytes(username, encoding='utf-8')) m.update(bytes(SECRET_KEY + str(time.time()), encoding='utf-8')) return m.hexdigest() # 用户登录接口
class LoginView(APIView): def post(self, request, *args, **kwargs): try: username = request.data["username"] password = request.data["password"] user_obj = models.UserInfo.objects.filter(username=username, password=password).first() if user_obj: # 为登录用户创建token
                token = md5(username) # 保存(存在就更新不存在就创建，并设置过期时间为5分钟)
                expiration_time = datetime.now() + dateutil.relativedelta.relativedelta(minutes=5) print(expiration_time, type(expiration_time)) defaults = { "token": token, "expiration_time": expiration_time } models.UserToken.objects.update_or_create(user=user_obj, defaults=defaults) return Response({"code": 200, "token": token}) else: return Response({"code": 401, "error": "用户名或密码错误"}) except Exception as e: print(e) return Response({"code": 500, "error": "用户名或密码错误"}) # 商品接口
class GoodsView(APIView): authentication_classes = [TokenAuthtication, ] def get(self, request, *args, **kwargs): print(request.user) print(request.auth) try: goods_data_list = [ {'id': 1, 'goods_name': "草莓", 'price': 19.99, 'status': True}, {'id': 2, 'goods_name': "香蕉", 'price': 9.88, 'status': True}, {'id': 3, 'goods_name': "苹果", 'price': 5.99, 'status': True}, {'id': 4, 'goods_name': "蓝莓", 'price': 9.99, 'status': True}, ] return Response({"code": 200, "msg": "商品接口", "data_list": goods_data_list}) except Exception as e: print(e) return Response({"code": 500, "error": "接口维护中..."}) # 用户信息接口
class UserInfoView(APIView): authentication_classes = [TokenAuthtication, ] def get(self, request, *args, **kwargs): try: username = request.user user_obj = models.UserInfo.objects.filter(username=username).first() data = {} data["username"] = user_obj.username data["user_type"] = user_obj.user_type data["add_time"] = user_obj.add_time return Response({"code": 200, "msg": "用户信息接口", "userinfo": data}) except Exception as e: print(e) return Response({"code": 500, "error": "接口维护中..."})

2.5 测试

http://127.0.0.1:8000/api/v1/login/

http://127.0.0.1:8000/api/v1/goods/?token=604127770384e4e067191f7ebe9f4fb0

http://127.0.0.1:8000/api/v1/userinfo/?token=e488b5948132c071335dd2f2bda14bb4

2.6 全局配置

################################################### models.py ################################################
from django.db import models
from datetime import datetime


# 用户模型
class UserInfo(models.Model):
    user_type_choics = (
        (1, "普通用户"),
        (2, "普通会员"),
        (3, "白金会员"),
        (4, "黄金会员"),
    )
    username = models.CharField(max_length=64, verbose_name="用户名")
    password = models.CharField(max_length=64, verbose_name="密码")
    user_type = models.IntegerField(choices=user_type_choics)
    add_time = models.DateTimeField(default=datetime.now, verbose_name="注册时间")

    class Mate:
        managed = True
        db_table = "user_info"
        verbose_name = "用户模型"
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.username


# 存放用户登录成功后的token
class UserToken(models.Model):
    user = models.OneToOneField(to='UserInfo', unique=True, on_delete=False)
    token = models.CharField(max_length=64, verbose_name="用户token")
    expiration_time = models.DateTimeField(default=datetime.now, verbose_name="过期时间")
    add_time = models.DateTimeField(default=datetime.now, verbose_name="添加时间")

    class Mate:
        managed = True
        db_table = "user_token"
        verbose_name = "用户Token"
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.token

############################################## urls.py ######################################################
from django.conf.urls import url
from . import views

urlpatterns = [
    # 用户登录接口
    url(r'login/$',views.LoginView.as_view()),

    # 商品列表接口
    url(r'goods/$',views.GoodsView.as_view()),

    # 用户信息接口
    url(r'userinfo/$',views.UserInfoView.as_view()),
]


################################################# views.py #################################################
import hashlib
from datetime import datetime
import dateutil.relativedelta
import time
from rest_framework.request import Request
from rest_framework.views import APIView
from rest_framework.response import Response
from . import models
from token_authentication.settings import SECRET_KEY

# 生成token
def md5(username):
    m = hashlib.md5(bytes(username, encoding='utf-8'))
    m.update(bytes(SECRET_KEY + str(time.time()), encoding='utf-8'))
    return m.hexdigest()

# 用户登录接口
class LoginView(APIView):
    # []表示不需要认证
    authentication_classes = []
    def post(self, request, *args, **kwargs):
        try:
            username = request.data["username"]
            password = request.data["password"]
            user_obj = models.UserInfo.objects.filter(username=username, password=password).first()
            if user_obj:
                # 为登录用户创建token
                token = md5(username)
                # 保存(存在就更新不存在就创建)
                expiration_time = datetime.now() + dateutil.relativedelta.relativedelta(minutes=5)
                print(expiration_time, type(expiration_time))
                defaults = {
                    "token": token,
                    "expiration_time": expiration_time
                }
                models.UserToken.objects.update_or_create(user=user_obj, defaults=defaults)
                return Response({"code": 200, "token": token})
            else:
                return Response({"code": 401, "error": "用户名或密码错误"})
        except Exception as e:
            print(e)
            return Response({"code": 500, "error": "用户名或密码错误"})


# 商品接口
class GoodsView(APIView):
    def get(self, request, *args, **kwargs):
        print(request.user)
        print(request.auth)
        try:
            goods_data_list = [
                {'id': 1, 'goods_name': "草莓", 'price': 19.99, 'status': True},
                {'id': 2, 'goods_name': "香蕉", 'price': 9.88, 'status': True},
                {'id': 3, 'goods_name': "苹果", 'price': 5.99, 'status': True},
                {'id': 4, 'goods_name': "蓝莓", 'price': 9.99, 'status': True},
            ]
            return Response({"code": 200, "msg": "商品接口", "data_list": goods_data_list})
        except Exception as e:
            print(e)
            return Response({"code": 500, "error": "接口维护中..."})


# 用户信息接口
class UserInfoView(APIView):
    def get(self, request, *args, **kwargs):
        try:
            username = request.user
            user_obj = models.UserInfo.objects.filter(username=username).first()
            data = {}
            data["username"] = user_obj.username
            data["user_type"] = user_obj.user_type
            data["add_time"] = user_obj.add_time
            return Response({"code": 200, "msg": "用户信息接口", "userinfo": data})
        except Exception as e:
            print(e)
            return Response({"code": 500, "error": "接口维护中..."})


############################################ token_auth.py ###############################################
from datetime import datetime
from rest_framework import exceptions
from rest_framework.authentication import BaseAuthentication
from api import models


# 基于token的用户认证
class TokenAuthtication(BaseAuthentication):
    def authenticate(self, request):
        # 1. 在请求头的query_params中获取token
        # token = request.query_params.get('token')

        # 2. 直接在请求头中获取token
        token = request._request.GET.get('token')
        token_obj = models.UserToken.objects.filter(token=token).first()
        if not token_obj:
            raise exceptions.AuthenticationFailed("用户认证失败")
        else:
            datetime_now = datetime.now()
            if token_obj.expiration_time > datetime_now:
                # 在 rest framework 内部会将两个字段赋值给request，以供后续操作使用
                return (token_obj.user, token_obj)
            else:
                raise exceptions.AuthenticationFailed("用户token过期,请重新登录")

    def authenticate_header(self, request):
        # 验证失败时，返回的响应头WWW-Authenticate对应的值
        pass


################################################## settings.py #######################################################################
# django restframework 配置
REST_FRAMEWORK = {
    # 用户认证
    "DEFAULT_AUTHENTICATION_CLASSES" : ['api.utils.token_auth.TokenAuthtication',]
}