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

朴灿烈づ我的快乐病毒、 2023-10-10 13:13 48阅读 0赞

## 功能简介 ##

业务需求：今天想尝试为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')

效果展示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3V6aGl3ZW5nYW5n_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3V6aGl3ZW5nYW5n_size_16_color_FFFFFF_t_70 1][]

项目源码地址：

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3V6aGl3ZW5nYW5n_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210811180359459.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3V6aGl3ZW5nYW5n,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3V6aGl3ZW5nYW5n_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210811182434279.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3V6aGl3ZW5nYW5n,size_16,color_FFFFFF,t_70