Django前后端分离restframework

基本配置

安装

pip install djangorestframework

配置

#允许访问的ip，*代表所有
ALLOWED_HOSTS = ["*"]
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    # 注册drf
    'rest_framework',
]

URL 和视图

# 类继承的视图
from rest_framework.views import APIView
# 返回数据方法
from rest_framework.response import Response

手动去除csrf_token,加上装饰器csrf_exempt

from django.views.decorators.csrf import csrf_exempt
@csrf_exempt
def user(request):
    pass

request 的取值

request.headers.get('token')
request.COOKIES.get("token")
request.query_params.get("token")
request.data.get("xxx")

drf版本控制

参数形式

路由url

from django.urls import path
from app01 import views
urlpatterns = [
    path('api/users/',views.UserView.as_view()),
]

视图views

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.versioning import QueryParameterVersioning


# Create your views here.
class UserView(APIView):
    versioning_class = QueryParameterVersioning

    def get(self, request, *args, **kwargs):
        print(request.version)
        return Response({"code": 200, "data": "ok"})

1.获取固定 version 版本号

访问地址：http://127.0.0.1:8000/api/users/?version=v1

2.自定义版本参数

自定义访问地址：http://127.0.0.1:8000/api/users/?v=v123

默认访问地址：http://127.0.0.1:8000/api/users

REST_FRAMEWORK = {
    # 自定义参数获取版本号
    "VERSION_PARAM": "v",
    # 未传入版本的默认版本
    "DEFAULT_VERSION": "v2",
}

3.允许的版本号范围

REST_FRAMEWORK = {
    # 自定义参数获取版本号
    "VERSION_PARAM": "v",
    # 未传入版本的默认版本
    "DEFAULT_VERSION": "v2",
    # 允许的版本号
    "ALLOWED_VERSIONS": ["v1", "v2", "v3"]
}

4.全局设置处理版本信息

REST_FRAMEWORK = {
    # 自定义参数获取版本号
    "VERSION_PARAM": "v",
    # 未传入版本的默认版本
    "DEFAULT_VERSION": "v2",
    # 允许的版本号
    "ALLOWED_VERSIONS": ["v1", "v2", "v3"],
    # 全局配置处理版本路径,所有视图都不需要加versioning_class = QueryParameterVersioning
    "DEFAULT_VERSIONING_CLASS":"rest_framework.versioning.QueryParameterVersioning"
}

路径形式

路由

urlpatterns = [
    # 全局版本取值参数<str:version>
    path('api/<str:version>/users/',views.UserView.as_view()),
]

视图

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.versioning import URLPathVersioning


# Create your views here.
class UserView(APIView):
    # 单独配置
    versioning_class = URLPathVersioning

    def get(self, request, *args, **kwargs):
        print(request.version)
        return Response({"code": 200, "data": "ok"})

配置

REST_FRAMEWORK = {
    # 路由的获取参数值
    "VERSION_PARAM": "version",
    # 允许的列表
    "ALLOWED_VERSIONS": ["v1", "v2", "v3"],
    # 全局配置
    "DEFAULT_VERSIONING_CLASS": "rest_framework.versioning.URLPathVersioning",
}

请求头形式

url.py

from django.urls import path
from app01 import views
urlpatterns = [
    path('api/users/', views.UserView.as_view()),
]

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.versioning import AcceptHeaderVersioning


# Create your views here.
class UserView(APIView):
    versioning_class = AcceptHeaderVersioning

    def get(self, request, *args, **kwargs):
        print(request.version)
        return Response({"code": 200, "data": "get"})

    def post(self, request, *args, **kwargs):
        return Response({"code": 200, "data": "post"})

二级域名

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.versioning import HostNameVersioning

路由 namespaces

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.versioning import NamespaceVersioning

反向生成 URL

urls.py

from django.urls import path
from app01 import views
urlpatterns = [
    path('api/<str:version>/users/', views.UserView.as_view(),name="v2"),
]

views.py

class UserView(APIView):

    def get(self, request, *args, **kwargs):
        print(request.version)
        # v2是路由传的name一一对应
        url1=request.versioning_scheme.reverse("v2",request=request)
        print(url1)
        return Response({"code": 200, "data": "get"})

    def post(self, request, *args, **kwargs):
        return Response({"code": 200, "data": "post"})

认证组件

在视图类的 authentication_classes 中定义认证类时，传入的是一个列表，支持定义多个认证类。

当出现多个认证类时，drf 内部会按照列表的顺序，逐一执行认证类的 authenticate 方法，如果 返回元组 或 抛出异常 则会终止后续认证类的执行；如果返回 None，则意味着继续执行后续的认证类。

如果所有的认证类authenticate都返回了 None，则默认 request.user="AnonymousUser" 和 request.auth=None，也可以通过修改配置文件来修改默认值。

更改返回为 None

REST_FRAMEWORK = {
 "UNAUTHENTICATED_USER": lambda: None,
 "UNAUTHENTICATED_TOKEN": lambda: None,
}

”返回 None“的应用场景：

当某个API，已认证 和 未认证 的用户都可以方法时，比如：

• 已认证用户，访问API返回该用户的视频播放记录列表。
• 未认证用户，访问API返回最新的的视频列表。

注意：不同于之前的案例，之前案例是：必须认证成功后才能访问，而此案例则是已认证和未认证均可访问。

单独认证类

views.py

import uuid
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from app01 import models


# 用户登录
class AuthView(APIView):
    def post(self, request, *args, **kwargs):
        print(request.data)
        username = request.data.get("username")
        password = request.data.get("password")
        print(username, password)
        user_object = models.UserInfo.objects.filter(username=username, password=password).first()
        print(user_object)
        if not user_object:
            return Response({"code": 1000, "data": "用户名或密码错误"})

        user_token = str(uuid.uuid4())
        user_object.token = user_token
        user_object.save()
        return Response({"code": 0, "data": {"token": user_token, "name": username}})


# 认证组件
class TokenAuthentication(BaseAuthentication):
    def authenticate(self, request):
        token = request.query_params.get("token")
        if not token:
            raise AuthenticationFailed({"code": 1002, "data": "认证失败"})
        user_object = models.UserInfo.objects.filter(token=token).first()
        if not user_object:
            raise AuthenticationFailed({"code": 1002, "data": "认证失败"})
        return user_object, token

    def authenticate_header(self, request):
        return 'Bearer realm="API"'


# 订单
class OrderView(APIView):
    # 增加认证
    authentication_classes = [TokenAuthentication, ]

    def get(self, request, *args, **kwargs):
        user=request.user
        print(type(user))
        print(request.auth)
        return Response({"code": 1000, "data": {"user": request.user.username, "list": [1, 2, 3, 4, 5, 6]}})

    def post(self, request, *args, **kwargs):
        pass


# 支付
class PayView(APIView):
    authentication_classes = [TokenAuthentication, ]

    def get(self, request, *args, **kwargs):
        pass

urls.py

urlpatterns = [
    path('auth/', views.AuthView.as_view()),
    path('order/', views.OrderView.as_view()),
    path('pay/', views.PayView.as_view()),
]

models.py

from django.db import models


class UserInfo(models.Model):
    username = models.CharField(verbose_name="用户名", max_length=32)
    password = models.CharField(verbose_name="密码", max_length=64)
    token = models.CharField(verbose_name="TOKEN", max_length=64, null=True, blank=True)

    def __str__(self):
        return self.username

关于多个认证类

一般情况下，编写一个认证类足矣。

当项目中可能存在多种认证方式时，就可以写多个认证类。例如，项目认证支持：

• 在请求中传递 token 进行验证。
• 请求携带 cookie 进行验证。
• 请求携带jwt进行验证。
• 请求携带的加密的数据，需用特定算法解密（一般为app开发的接口都是有加密算法）

auth.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# author:隆海 | www.pythl.com
# datetime:2022/6/25 11:17
# software: PyCharm
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from app01 import models


# Token认证组件
class TokenAuthentication(BaseAuthentication):
    def authenticate(self, request):
        token = request.query_params.get("token")
        if not token:
            return None
        user_object = models.UserInfo.objects.filter(token=token).first()
        if not user_object:
            raise AuthenticationFailed({"code": 1002, "data": "认证失败"})
        assert isinstance(token, object)
        return user_object, token

    def authenticate_header(self, request):
        return 'Bearer realm="API"'


# Cookie认证组件
class CookieAuthentication(BaseAuthentication):
    def authenticate(self, request):
        cookie_id = request.COOKIES.get("token")
        if not cookie_id:
            return None
        user_object = models.UserInfo.objects.filter(cookie_id=cookie_id).first()
        if not user_object:
            raise AuthenticationFailed({"code": 1002, "data": "认证失败"})
        return user_object, cookie_id

    def authenticate_header(self, request):
        return 'Bearer realm="API"'

全局配置

REST_FRAMEWORK = {
    "VERSION_PARAM": "version",
    "ALLOWED_VERSIONS": ["v1", "v2", "v3"],
    "DEFAULT_VERSIONING_CLASS": "rest_framework.versioning.URLPathVersioning",
    "UNAUTHENTICATED_USER": lambda: None,
    "UNAUTHENTICATED_TOKEN": lambda: None,
    "DEFAULT_AUTHENTICATION_CLASSES":["app01.auth.TokenAuthentication","app01.auth.CookieAuthentication"],
}

自定义用户模型

我们自定义的用户模型类还不能直接被 Django 的认证系统所识别，需要在配置文件中告知 Django 认证系统使用我们自定义的模型类。

在 settings/dev.py 配置文件中进行设置

AUTH_USER_MODEL = 'users.User'

AUTH_USER_MODEL 参数的设置以点.来分隔，表示应用名.模型类名。

定义用户模型

from django.db import models
from django.contrib.auth.models import AbstractUser


class UserView(AbstractUser):
    """用户信息"""
    nickname = models.CharField(max_length=50, default="", null=True, verbose_name="用户昵称")
    mobile = models.CharField(max_length=11, unique=True, verbose_name='手机号', help_text='手机号',
                              error_messages={'unique': "此手机号码已经注册"})
    avatar = models.ImageField(upload_to="avatar/%Y", null=True, default="", verbose_name="个人头像")
    sex_choice = (
        (0, "女"),
        (1, "男"),
    )
    sex = models.IntegerField(choices=sex_choice, default=1, verbose_name="性别")
    appid = models.CharField(max_length=30, default=0, verbose_name="微信appid")

    class Meta:
        db_table = 'users_info'
        verbose_name = '用户信息'
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.nickname


class UserDetail(models.Model):
    """用户详情"""
    uid = models.ForeignKey(UserView, verbose_name="用户信息", on_delete=models.CASCADE)
    money = models.DecimalField(max_digits=9, default=0.0, decimal_places=2, verbose_name="钱包余额")
    credit = models.IntegerField(default=0, verbose_name="积分")
    company = models.CharField(verbose_name='公司', max_length=100, default='暂无信息')
    introduce = models.TextField(verbose_name='简介', default='暂无介绍')
    profession = models.CharField(verbose_name='职业', max_length=100, default='暂无信息')
    wx = models.CharField(verbose_name='微信', max_length=50, default='暂无信息')
    qq = models.CharField(verbose_name='QQ', max_length=50, default='暂无信息')
    wb = models.CharField(verbose_name='微博', max_length=100, default='暂无信息')

    class Meta:
        db_table = 'users_detail'
        verbose_name = '用户详情'
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.wx

创建超级用户

(.venv) ➜  drf3.2 python manage.py createsuperuser
Username: xixi
Email address:
Password:
Password (again):
The password is too similar to the username.
This password is too short. It must contain at least 8 characters.
Bypass password validation and create user anyway? [y/N]: y
Superuser created successfully.

注意：Django 建议我们对于 AUTH_USER_MODEL 参数的设置一定要在第一次数据库迁移之前就设置好，否则后续使用可能出现未知错误。

这是表示有一个叫 admin 的子应用使用了原来的废弃的 auth.User 模型，但是目前数据库已经设置了默认的子应用为users的模型了，所以产生了冲突。那么这种冲突，我们需要重置下原来的 auth 模块的迁移操作，再次迁移就可以解决了。

解决步骤：
1. 备份数据库[如果刚开始开发，无需备份。]
   cd /Users/wslh/development/drf3.2
   mysqldump -uroot -p123 blog > 11_06_blog.sql

2. 注释掉users.User代码以及AUTH_USER_MODEL配置项，然后执行数据迁移回滚操作，把冲突的所有表迁移记录全部归零
   cd /Users/wslh/development/drf3.2
   # python manage.py migrate <子应用目录> zero  重置用户模型数据
   python manage.py migrate auth zero

3. 恢复users.User代码以及AUTH_USER_MODEL配置项，执行数据迁移。
   python manage.py makemigrations
   python manage.py migrate
4. 创建管理员查看auth功能是否能正常使用。
   python manage.py createsuperuser

认识 JWT

Json web token (JWT), 是为了在网络应用环境间传递声明而执行的一种基于 JSON 的开放标准（(RFC 7519).该 token 被设计为紧凑且安全的，特别适用于分布式站点的单点登录（SSO）场景。JWT 的声明一般被用来在身份提供者（客户端）和服务提供者（服务端）间传递被认证的用户身份信息，以便于从资源服务器获取资源，也可以增加一些额外的其它业务逻辑所必须的声明信息，该 token 也可直接被用于身份认证，也可被数据加密传输。

JWT 的构成

JWT 就一段字符串，由三段信息构成的，将这三段信息文本用.拼接一起就构成了 Jwt token 字符串。就像这样:

eyJ0eXAiOiAiand0IiwgImFsZyI6ICJIUzI1NiJ9.eyJzdWIiOiAicm9vdCIsICJleHAiOiAiMTUwMTIzNDU1IiwgImlhdCI6ICIxNTAxMDM0NTUiLCAibmFtZSI6ICJ3YW5neGlhb21pbmciLCAiYWRtaW4iOiB0cnVlLCAiYWNjX3B3ZCI6ICJRaUxDSmhiR2NpT2lKSVV6STFOaUo5UWlMQ0poYkdjaU9pSklVekkxTmlKOVFpTENKaGJHY2lPaUpJVXpJMU5pSjkifQ==.815ce0e4e15fff813c5c9b66cfc3791c35745349f68530bc862f7f63c9553f4b

第一部分我们称它为头部（header)，第二部分我们称其为载荷（payload, 类似于飞机上承载的物品)，第三部分是签证（signature).

header

jwt 的头部承载两部分信息：

• typ: 声明 token 类型，这里是 jwt ，typ 的值也可以是：Bear
• alg: 声明签证的加密的算法 通常直接使用 HMAC SHA256

完整的头部就像下面这样的 JSON：

{
  'typ': 'JWT',
  'alg': 'HS256'
}

然后将头部进行 base64 编码，构成了 jwt 的第一部分头部

python 代码举例：

import base64, json
header_data = {"typ": "jwt", "alg": "HS256"}
header = base64.b64encode( json.dumps(header_data).encode() ).decode()
print(header) # eyJ0eXAiOiAiand0IiwgImFsZyI6ICJIUzI1NiJ9

payload

载荷就是存放有效信息的地方。这个名字像是特指飞机上承载的货仓，这些有效信息包含三个部分:

• 标准声明
• 公共声明
• 私有声明

标准声明指定 jwt 实现规范中要求的属性。 (官方建议但不强制使用) ：

• iss: jwt 签发者
• sub: jwt 所面向的用户
• aud: 接收 jwt 的一方
• exp: jwt 的过期时间，这个过期时间必须要大于签发时间
• nbf: 定义在什么时间之后，该 jwt 才可以使用
• iat: jwt 的签发时间
• jti: jwt 的唯一身份标识，主要用来作为一次性 token, 从而回避重放攻击。

公共声明 ： 公共的声明可以添加任何的公开信息，一般添加用户的相关信息或其他业务需要的必要信息.但不建议添加敏感信息，因为该部分在客户端可直接读取.

私有声明 ： 私有声明是提供者和消费者所共同定义的声明，一般不建议存放敏感信息，里面存放的是一些可以在服务端或者客户端通过秘钥进行加密和解密的加密信息。往往采用的 RSA 非对称加密算法。

举例，定义一个 payload 载荷信息，demo/jwtdemo.py：

import base64, json, time

if __name__ == '__main__':
    # 载荷
    iat = int(time.time())
    payload_data = {
        "sub": "root",
        "exp": iat + 3600,  # 假设一小时过期
        "iat": iat,
        "name": "wangxiaoming",
        "avatar": "1.png",
        "user_id": 1,
        "admin": True,
        "acc_pwd": "QiLCJhbGciOiJIUzI1NiJ9QiLCJhbGciOiJIUzI1NiJ9QiLCJhbGciOiJIUzI1NiJ9",
    }
    # 将其进行base64编码，得到JWT的第二部分。
    payload = base64.b64encode(json.dumps(payload_data).encode()).decode()
    print(payload)
    # eyJzdWIiOiAicm9vdCIsICJleHAiOiAxNjQ3Nzc0Mjk1LCAiaWF0IjogMTY0Nzc3MDY5NSwgIm5hbWUiOiAid2FuZ3hpYW9taW5nIiwgImF2YXRhciI6ICIxLnBuZyIsICJ1c2VyX2lkIjogMSwgImFkbWluIjogdHJ1ZSwgImFjY19wd2QiOiAiUWlMQ0poYkdjaU9pSklVekkxTmlKOVFpTENKaGJHY2lPaUpJVXpJMU5pSjlRaUxDSmhiR2NpT2lKSVV6STFOaUo5In0=

signature

JWT 的第三部分是一个签证信息，用于辨真伪，防篡改。这个签证信息由三部分组成：

• header (base64 后的头部)
• payload (base64 后的载荷)
• secret（保存在服务端的秘钥字符串，不会提供给客户端的，这样可以保证客户端没有签发 token 的能力）

举例，定义一个完整的 jwt token，demo/jwtdemo.py：

import base64, json, hashlib

if __name__ == '__main__':
    """jwt 头部的生成"""
    header_data = {"typ": "jwt", "alg": "HS256"}
    header = base64.b64encode( json.dumps(header_data).encode() ).decode()
    print(header) # eyJ0eXAiOiAiand0IiwgImFsZyI6ICJIUzI1NiJ9

    """jwt 载荷的生成"""
    payload_data = {
        "sub": "root",
        "exp": "150123455",
        "iat": "150103455",
        "name": "wangxiaoming",
        "admin": True,
        "acc_pwd": "QiLCJhbGciOiJIUzI1NiJ9QiLCJhbGciOiJIUzI1NiJ9QiLCJhbGciOiJIUzI1NiJ9",
    }
    # 将其进行base64编码，得到JWT的第二部分。
    payload = base64.b64encode(json.dumps(payload_data).encode()).decode()
    print(payload) # eyJzdWIiOiAicm9vdCIsICJleHAiOiAiMTUwMTIzNDU1IiwgImlhdCI6ICIxNTAxMDM0NTUiLCAibmFtZSI6ICJ3YW5neGlhb21pbmciLCAiYWRtaW4iOiB0cnVlLCAiYWNjX3B3ZCI6ICJRaUxDSmhiR2NpT2lKSVV6STFOaUo5UWlMQ0poYkdjaU9pSklVekkxTmlKOVFpTENKaGJHY2lPaUpJVXpJMU5pSjkifQ==

    # from django.conf import settings
    # secret = settings.SECRET_KEY
    secret = 'django-insecure-hbcv-y9ux0&8qhtkgmh1skvw#v7ru%t(z-#chw#9g5x1r3z=$p'
    data = header + payload + secret  # 秘钥绝对不能提供给客户端。
    HS256 = hashlib.sha256()
    HS256.update(data.encode('utf-8'))
    signature = HS256.hexdigest()
    print(signature) # 815ce0e4e15fff813c5c9b66cfc3791c35745349f68530bc862f7f63c9553f4b

    # jwt 最终的生成
    token = f"{header}.{payload}.{signature}"
    print(token)
    # eyJ0eXAiOiAiand0IiwgImFsZyI6ICJIUzI1NiJ9.eyJzdWIiOiAicm9vdCIsICJleHAiOiAiMTUwMTIzNDU1IiwgImlhdCI6ICIxNTAxMDM0NTUiLCAibmFtZSI6ICJ3YW5neGlhb21pbmciLCAiYWRtaW4iOiB0cnVlLCAiYWNjX3B3ZCI6ICJRaUxDSmhiR2NpT2lKSVV6STFOaUo5UWlMQ0poYkdjaU9pSklVekkxTmlKOVFpTENKaGJHY2lPaUpJVXpJMU5pSjkifQ==.815ce0e4e15fff813c5c9b66cfc3791c35745349f68530bc862f7f63c9553f4b

注意：secret 是保存在服务器端的，jwt 的签发生成也是在服务器端的，secret 就是用来进行 jwt 的签发和 jwt 的验证，所以，它就是你服务端的私钥，在任何场景都不应该流露出去。一旦客户端得知这个 secret, 那就意味着客户端是可以自我签发 jwt 了。

定义一个完整的 jwt token，并认证 token，demo/jwtdemo.py：

import base64, json, hashlib
from datetime import datetime

if __name__ == '__main__':
    # 头部生成原理
    header_data = {
        "typ": "jwt",
        "alg": "HS256"
    }
    # print( json.dumps(header_data).encode() )
    # json转成字符串，接着base64编码处理
    header = base64.b64encode(json.dumps(header_data).encode()).decode()
    print(header)  # eyJ0eXAiOiAiand0IiwgImFsZyI6ICJIUzI1NiJ9


    # 载荷生成原理
    iat = int(datetime.now().timestamp()) # 签发时间
    payload_data = {
        "sub": "root",
        "exp": iat + 3600,  # 假设一小时过期
        "iat": iat,
        "name": "wangxiaoming",
        "admin": True,
        "acc_pwd": "QiLCJhbGciOiJIUzI1NiJ9QiLCJhbGciOiJIUzI1NiJ9QiLCJhbGciOiJIUzI1NiJ9",
    }

    payload = base64.b64encode(json.dumps(payload_data).encode()).decode()
    print(payload)
    # eyJzdWIiOiAicm9vdCIsICJleHAiOiAxNjM2NTk3OTAzLCAiaWF0IjogMTYzNjU5NDMwMywgIm5hbWUiOiAid2FuZ3hpYW9taW5nIiwgImFkbWluIjogdHJ1ZSwgImFjY19wd2QiOiAiUWlMQ0poYkdjaU9pSklVekkxTmlKOVFpTENKaGJHY2lPaUpJVXpJMU5pSjlRaUxDSmhiR2NpT2lKSVV6STFOaUo5In0=

    # from django.conf import settings
    # secret = settings.SECRET_KEY
    secret = 'django-insecure-hbcv-y9ux0&8qhtkgmh1skvw#v7ru%t(z-#chw#9g5x1r3z=$p'

    data = header + payload + secret  # 秘钥绝对不能提供给客户端。

    HS256 = hashlib.sha256()
    HS256.update(data.encode('utf-8'))
    signature = HS256.hexdigest()
    print(signature) # ce46f9d350be6b72287beb4f5f9b1bc4c42fc1a1f8c8db006e9e99fd46961156

    # jwt 最终的生成
    token = f"{header}.{payload}.{signature}"
    print(token)
    # eyJ0eXAiOiAiand0IiwgImFsZyI6ICJIUzI1NiJ9.eyJzdWIiOiAicm9vdCIsICJleHAiOiAxNjM2NTk3OTAzLCAiaWF0IjogMTYzNjU5NDMwMywgIm5hbWUiOiAid2FuZ3hpYW9taW5nIiwgImFkbWluIjogdHJ1ZSwgImFjY19wd2QiOiAiUWlMQ0poYkdjaU9pSklVekkxTmlKOVFpTENKaGJHY2lPaUpJVXpJMU5pSjlRaUxDSmhiR2NpT2lKSVV6STFOaUo5In0=.ce46f9d350be6b72287beb4f5f9b1bc4c42fc1a1f8c8db006e9e99fd46961156


    # 认证环节
    token = "eyJ0eXAiOiAiand0IiwgImFsZyI6ICJIUzI1NiJ9.eyJzdWIiOiAicm9vdCIsICJleHAiOiAxNjM2NTk3OTAzLCAiaWF0IjogMTYzNjU5NDMwMywgIm5hbWUiOiAid2FuZ3hpYW9taW5nIiwgImFkbWluIjogdHJ1ZSwgImFjY19wd2QiOiAiUWlMQ0poYkdjaU9pSklVekkxTmlKOVFpTENKaGJHY2lPaUpJVXpJMU5pSjlRaUxDSmhiR2NpT2lKSVV6STFOaUo5In0=.ce46f9d350be6b72287beb4f5f9b1bc4c42fc1a1f8c8db006e9e99fd46961156"
    # token = "eyJ0eXAiOiAiand0IiwgImFsZyI6ICJIUzI1NiJ9.eyJzdWIiOiJyb290IiwiZXhwIjoxNjMxNTI5MDg4LCJpYXQiOjE2MzE1MjU0ODgsIm5hbWUiOiJ3YW5neGlhb2hvbmciLCJhZG1pbiI6dHJ1ZSwiYWNjX3B3ZCI6IlFpTENKaGJHY2lPaUpJVXpJMU5pSjlRaUxDSmhiR2NpT2lKSVV6STFOaUo5UWlMQ0poYkdjaU9pSklVekkxTmlKOSJ9.b533c5515444c51058557017e433d411379862d91640c8beed6f2617b1da2feb"
    header, payload, signature = token.split(".")

    # 验证是否过期了
    # 先基于base64，接着使用json解码
    payload_data = json.loads( base64.b64decode(payload.encode()) )
    print(payload_data)
    exp = payload_data.get("exp", None)
    if exp is not None and int(exp) < int(datetime.now().timestamp()):
        print("token过期！！！")
    else:
        print("没有过期")

    # 验证token是否有效，是否被篡改
    # from django.conf import settings
    # secret = settings.SECRET_KEY
    secret = 'django-insecure-hbcv-y9ux0&8qhtkgmh1skvw#v7ru%t(z-#chw#9g5x1r3z=$p'
    data = header + payload + secret  # 秘钥绝对不能提供给客户端。
    HS256 = hashlib.sha256()
    HS256.update(data.encode('utf-8'))
    new_signature = HS256.hexdigest()

    if new_signature != signature:
        print("认证失败")
    else:
        print("认证通过")

关于签发和核验 JWT，python 中提供了一个 PyJWT 模块帮我们实现 jwt 的整体流程。我们可以使用 Django REST framework JWT 扩展来完成。

4.0 之前文档网站：https://jpadilla.github.io/django-rest-framework-jwt/

4.0 以后需要使用 simplejwt:https://django-rest-framework-simplejwt.readthedocs.io/en/latest/

安装配置 JWT

安装

pip install djangorestframework-jwt

settings/dev.py，配置 jwt

# drf配置
REST_FRAMEWORK = {
    # 自定义异常处理
    'EXCEPTION_HANDLER': 'luffycityapi.utils.exceptions.exception_handler',
    # 自定义认证
    'DEFAULT_AUTHENTICATION_CLASSES': (
        'rest_framework_jwt.authentication.JSONWebTokenAuthentication',  # jwt认证
        'rest_framework.authentication.SessionAuthentication',           # session认证
        'rest_framework.authentication.BasicAuthentication',
    ),
}

import datetime
# jwt认证相关配置项
JWT_AUTH = {
    # 设置jwt的有效期
    # 如果内部站点，例如：运维开发系统，OA，往往配置的access_token有效期基本就是15分钟，30分钟，1~2个小时
    'JWT_EXPIRATION_DELTA': datetime.timedelta(weeks=1), # 一周有效，
}

• JWT_EXPIRATION_DELTA 指明 token 的有效期

生成 jwt

Django REST framework JWT 扩展的说明文档中提供了手动签发 JWT 的方法

官方文档：https://jpadilla.github.io/django-rest-framework-jwt/#creating-a-new-token-manually

# 可以进入到django的终端下测试生成token的逻辑
python manage.py shell

# 引入jwt配置
from rest_framework_jwt.settings import api_settings
# 获取载荷生成函数
jwt_payload_handler = api_settings.JWT_PAYLOAD_HANDLER
# 获取token生成函数
jwt_encode_handler = api_settings.JWT_ENCODE_HANDLER
# 生成载荷需要的字典数据
# 此处，拿数据库中的用户信息进行测试
from users.models import User
user = User.objects.first()
payload = jwt_payload_handler(user)  # user用户模型对象
# 生成token
token = jwt_encode_handler(payload)

在用户注册或登录成功后，在序列化器中返回用户信息以后同时返回 token 即可。

权限组件

permissions.py

# 权限类
class PermissionA(BasePermission):
    message = {"code": 1000, "data": "无权访问"}

    def has_permission(self, request, view):
        # 权限对象是True为验证成功
        try:
            if request.user.role == 1:
                return True
        except Exception as e:
            return False
        return False

    def has_object_permission(self, request, view, obj):
        return True

views.py

# 订单
class OrderView(APIView):
    # 增加认证
    authentication_classes = [TokenAuthentication, ]
    permission_classes = [PermissionA, ]

    def get(self, request, *args, **kwargs):
        print(2, request.user)
        if not request.user:
            return Response(dict(code=1000, data={"user": None, "data": [1, 2, 3, 4, 5, 6]}))
        return Response({"code": 1000, "data": {"user": request.user.username, "list": [10, 11, 12, 13]}})

小结

• 请求的封装
• 版本的处理
  • 过程：选择版本处理类，获取用户传入的版本信息
  • 结果：在 request.version = 版本、request.versioning_scheme=版本处理类的对象
• 认证组件，在视图执行之前判断用户是否认证成功。
  • 过程：执行所有的认证类中的 authenticate 方法
    • 返回 None，继续执行后续的认证类（都未认证成功，request.user 和 auth 有默认值，也可以全局配置）
    • 返回 2 个元素的元组，中断
    • 抛出 AuthenticationFailed，中断
  • 结果：在 request.user 和 request.auth 赋值（后续代码可以使用）
• 权限
  • 过程：执行所有的权限类的has_permission方法，只有所有都返回 True 时，才表示具有权限
  • 结果：有权限则可以执行后续的视图，无权限则直接返回 自定义的错误信息

限流组件

限流，限制用户访问频率，例如：用户 1 分钟最多访问 100 次 或者 短信验证码一天每天可以发送 50 次， 防止盗刷。

• 对于匿名用户，使用用户 IP 作为唯一标识。
• 对于登录用户，使用用户 ID 或名称作为唯一标识。

settings.py

# 缓存redis配置
CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": "redis://127.0.0.1:6379",
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
            "PASSWORD": "",
        }
    }
}

throttles.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# author:隆海 | www.pythl.com
# datetime:2022/6/25 15:42
# software: PyCharm

from rest_framework.throttling import SimpleRateThrottle
from django.core.cache import cache as default_cache
from rest_framework import status
from rest_framework import exceptions


class ThrottledException(exceptions.APIException):
    status_code = status.HTTP_429_TOO_MANY_REQUESTS
    default_code = "throttled"


class MyRateThrottle(SimpleRateThrottle):
    cache = default_cache  # 访问记录存放在django的缓存中（需设置缓存）
    scope = "user"  # 构造缓存中的key
    cache_format = "throttle_%(scope)s_%(ident)s"

    # 设置访问频率，例如：1分钟允许访问10次
    # 其他：‘s'，'sec'，'m'，'min'，‘h'，‘hour'，'d'，‘day'
    THROTTLE_RATES = {"user": "5/m"}  # 10代表10次，m为分钟

    # 返回唯一标识
    def get_cache_key(self, request, view):
        if request.user:
            ident = request.user.pk  # 用户ID
        else:
            ident = self.get_ident(request)  # 获取请求用户IP(去request中找请求头）
        return self.cache_format % {"scope": self.scope, "ident": ident}

    # 超出访问时的提示信息,如果多个限流就不写此方法，在视图中书写
    def throttle_failure(self):
        wait = self.wait()
        detail = {
            "code": 1005,
            "data": "访问频率限制",
            "detail": "需等待{}s才能访问".format(int(wait))
        }
        raise ThrottledException(detail)

views.py

from rest_framework.views import APIView
from rest_framework.response import Response
from app01 import models
from app01.throttles import MyRateThrottle,ThrottledException


# 单个限流
class PayView(APIView):
    throttle_classes = [MyRateThrottle, ]

    def get(self, request, *args, **kwargs):
        if not request.user:
            return Response({"code": 1000, "data": {"user": None, "data": [1, 2, 3, 4, 5, 6]}})

        return Response({"code": 1000, "data": {"user": request.user.username, "list": [10, 11, 12, 13]}})

# 多个限流，限流类return True/False 返回信息在视图中重写，如果限流类报错就不会执行下一个限流类
class PayView(APIView):
    throttle_classes = [MyRateThrottleA, MyRateThrottleB]

    def get(self, request, *args, **kwargs):
        if not request.user:
            return Response({"code": 1000, "data": {"user": None, "data": [1, 2, 3, 4, 5, 6]}})

        return Response({"code": 1000, "data": {"user": request.user.username, "list": [10, 11, 12, 13]}})
      def throttled(self, request, wait):
        wait = self.wait()
        detail = {
            "code": 1005,
            "data": "访问频率限制",
            "detail": "需等待{}s才能访问".format(int(wait))
        }
        raise ThrottledException(detail)

全局配置

settings.py

REST_FRAMEWORK = {
    "DEFAULT_THROTTLE_CLASSES":["app01.throttles.MyRateThrottle"],
    "DEFAULT_THROTTLE_RATES":{
        "user":"10/m",
        "xx":"10/m"
    }
}

Serializer

drf 中为我们提供了 Serializer，他主要有两大功能：

• 对请求数据校验（底层调用Django的 Form 和ModelForm）
• 对数据库查询到的对象进行序列化

数据校验

序列化

数据校验&序列化