Django 集成JWT 基于Token凭证方式，实现前后端分离

Posted 2021-09-05 在奋斗的大道

功能简介

业务需求：今天想尝试为Django 添加安全验证框架，但是看了相关Django 涉及Authentication模块的配置，过于繁琐且不适合当前前后端分离趋势发展，所以放弃Django自带安全模块，使用Django 集成Jwt 基于Token 凭证方式，实现前后端功能分离。

业务思路大致步骤描述:

1、用户通过登入url,输入合法用户名和密码获取登入凭证token

2、后台业务逻辑方法通过装饰器函数(token_required())，校验请求接口携带的Token 是否合法、是否失效等。

3、token 凭证验证通过，执行后台业务逻辑相关方法，并返回结果。

功能实现

第一步：Django 项目添加JWT 模块

pip install pyjwt

第二步：在Django 应用之models.py 文件中添加用户认证实体对象(User)

class User(models.Model):
    id = models.BigAutoField(primary_key=True)
    name = models.CharField(max_length=255, null=True)
    passwd = models.CharField(max_length=255, null=True)

    @property
    def token(self):
        return self._generate_jwt_token()

    def _generate_jwt_token(self):
        token = jwt.encode({
            'exp': datetime.datetime.now() + datetime.timedelta(days=1),
            'iat': datetime.datetime.now(),
            'data': {
                'name': self.name
            }
        }, settings.SECRET_KEY, algorithm='HS256')

        return token.encode("utf-8").decode('utf-8')

    class Meta:
        managed = False
        db_table = 'user'

备注说明：声明一个私有属性方法_generate_jwt_token(self)=用于生成token 凭证 

                 声明一个实体属性token(self) = 用于获取token 凭证

第三步：在Django 应用之check.py 文件中添加权限校验装饰器函数token_required()，主要用于：校验token 合法性和用户合法性

核心代码:

# 登入凭证装饰器
def token_required():
    def decorator(view_func):
        def _wrapped_view(request, *args, **kwargs):
            try:
                auth = request.META.get('HTTP_AUTHORIZATION').split("_")
            except AttributeError:
                return response_failure("No authenticate header")

            if auth[0].lower() == 'token':
                try:
                    dict = jwt.decode(auth[1], settings.SECRET_KEY, algorithms=['HS256'])
                    username = dict.get('data').get('name')
                except jwt.ExpiredSignatureError:
                    return response_failure("Token expired")
                except jwt.InvalidTokenError:
                    return response_failure("Invalid token")
                except Exception as e:
                    return response_failure("Can not get user object")

                try:
                    user = User.objects.get(name=username)
                except User.DoesNotExist:
                    return response_failure("User Does not exist")

                return view_func(request, *args, **kwargs)
            else:
                return response_failure("Error authenticate header")

        return _wrapped_view

    return decorator

第四步：在Django 应用之views.py 文件中导入权限校验装饰器函数token_required(),并在需要校验token的方法前面，添加相关注解。

views.py 核心代码:

import json
from userSystem.models import Book
from userSystem.models import User
from django.http import HttpResponse
from django.core import serializers
from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger
from userSystem.check import request_verify
from userSystem.check import token_required
from userSystem.formValidation import BookFrom
from userSystem.tool import objDictTool
import logging

# 日志输出常量定义
logger = logging.getLogger('mylogger')


# 登入方法认证
def login(request):
    json_str = request.body
    json_str = json_str.decode()  # python3.6及以上不用这一句代码
    dict_data = json.loads(json_str)  # loads把str转换为dict，dumps把dict转换为str
    # 录入用户
    username = dict_data.get('username')
    # 录入密码
    password = dict_data.get('password')
    # 查询用户是否存在
    u = User.objects.filter(name=username, passwd=password)
    if u:
        # 生成用户登入凭证
        token = u[0].token
        return response_success(message='用户登入成功', data=token)
    else:
        return response_failure(message='用户登入失败')


# Create your views here.
@token_required()
@request_verify('get')
def select(request):
    books = Book.objects.all()
    for i in range(len(books)):
        print("主键：%s   值：%s" % (i + 1, books[i]))

    return response_success(message='后台响应成功', data_list=serializers.serialize("json", books))


@token_required()
@request_verify('post')
def selectAll(request):
    books = Book.objects.all()
    for i in range(len(books)):
        print("主键：%s   值：%s" % (i + 1, books[i]))
    return response_success(message='后台响应成功', data_list=serializers.serialize("json", books))


@token_required()
@request_verify('get', ['page', 'pageSize'])
def selectPage(request):
    # 当前页码
    page = request.GET.get('page')
    # 当前分页大小
    page_size = request.GET.get('pageSize')
    book_list = Book.objects.all()
    # django 分页实体对象
    paginator = Paginator(book_list, page_size)
    # 查询总记录数
    total = paginator.count
    try:
        # 执行分页查询
        books = paginator.page(page)
    except PageNotAnInteger:
        # 执行分页查询,默认指定页码
        books = paginator.page(1)
    except EmptyPage:
        # 执行分页查询,默认指定页码
        books = paginator.page(paginator.num_pages)
    return response_page_success(message='后台响应成功', data_list=serializers.serialize("json", books), total=total,
                                 page=page, pageSize=page_size)


@token_required()
@request_verify('post', ['page', 'pageSize'])
def selectPageAll(request):
    json_str = request.body
    json_str = json_str.decode()  # python3.6及以上不用这一句代码
    dict_data = json.loads(json_str)  # loads把str转换为dict，dumps把dict转换为str

    # 当前页码
    page = dict_data.get('page')
    # 当前分页大小
    page_size = dict_data.get('pageSize')
    book_list = Book.objects.all()
    # django 分页实体对象
    paginator = Paginator(book_list, page_size)
    # 查询总记录数
    total = paginator.count
    try:
        # 执行分页查询
        books = paginator.page(page)
    except PageNotAnInteger:
        # 执行分页查询,默认指定页码
        books = paginator.page(1)
    except EmptyPage:
        # 执行分页查询,默认指定页码
        books = paginator.page(paginator.num_pages)
    return response_page_success(message='后台响应成功', data_list=serializers.serialize("json", books), total=total,
                                 page=page, pageSize=page_size)


@token_required()
def insert(request):
    # 获取参数,放入表单校验
    book_form = BookFrom(request.POST)
    # 判断校验是否成功
    if book_form.is_valid():  # 验证成功
        name = book_form.cleaned_data.get("name")
        author = book_form.cleaned_data.get("author")
        print("名称：%s   作者：%s" % (name, author))
        return response_success(message="Django 实体表单验证成功")
    else:
        errorDict = book_form.errors
        for key, value in errorDict.items():
            print("属性：%s   错误信息：%s" % (key, value))
        return response_success(message="Django 实体表单验证失败")


# json 数据提交,并转换为实体，执行入库操作
@token_required()
def insertJSON(request):
    logger.info("post request body 请求数据提交")
    json_str = request.body
    json_str = json_str.decode()  # python3.6及以上不用这一句代码
    dict_data = json.loads(json_str)  # loads把str转换为dict，dumps把dict转换为str

    item = Book()
    objDictTool.to_obj(item, **dict_data)
    print("名称: {}, 价格: {},  作者: {}".format(item.name, item.price, item.author))
    # 执行数据库插入
    item.save()
    return response_success(message="数据入库成功")


def response_success(message, data=None, data_list=[]):
    return HttpResponse(json.dumps({
        'code': 200,  # code由前后端配合指定
        'message': message,  # 提示信息
        'data': data,  # 返回单个对象
        'dataList': data_list  # 返回对象数组
    }, ensure_ascii=False), 'application/json')


def response_failure(message):
    return HttpResponse(json.dumps({
        'code': 500,
        'message': message
    }, ensure_ascii=False), 'application/json')


def response_page_success(message, data=None, data_list=[], total=None, page=None, pageSize=None):
    return HttpResponse(json.dumps({
        'code': 200,  # code由前后端配合指定
        'message': message,  # 提示信息
        'data': data,  # 返回单个对象
        'dataList': data_list,  # 返回对象数组
        'total': total,  # 记录总数
        'page': page,  # 当前页面
        'pageSize': pageSize  # 当前页面分页大小
    }, ensure_ascii=False), 'application/json')

效果展示：

项目源码地址：

链接：https://pan.baidu.com/s/1_T0eKa60No-XRLFOpWFE2w 
提取码：1234