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',]
}