（1）源码入口git
rest_framework_jwt.views.ObtainJSONWebToken 的 父类 JSONWebTokenAPIView 的 post 方法
PS：只有post方法 接受username 与 password请求
（2）校验方式django
post方法将请求数据交给 rest_framework_jwt.serializer.JSONWebTokenSerializer 处理 PS 完成数据的校验，会走序列化类的 全局钩子校验规则，校验获得登陆用户并签发token存储在序列化对象中
（3）核心源码后端
def validate(self, attrs): # 帐号密码字典 credentials = { self.username_field: attrs.get(self.username_field), 'password': attrs.get('password') } if all(credentials.values()): # 签发token第1步：用帐号密码获得user对象 user = authenticate(**credentials) if user: if not user.is_active: msg = _('User account is disabled.') raise serializers.ValidationError(msg) # 签发token第2步：经过user获得payload，payload包含着用户信息与过时时间 payload = jwt_payload_handler(user) # 在视图类中，能够经过 序列化对象.object.get('user'或者'token') 拿到user和token return { # 签发token第3步：经过payload签发出token 'token': jwt_encode_handler(payload), 'user': user } else: msg = _('Unable to log in with provided credentials.') raise serializers.ValidationError(msg) else: msg = _('Must include "{username_field}" and "password".') msg = msg.format(username_field=self.username_field) raise serializers.ValidationError(msg)
（1）经过username password 生成user对象api
（2）经过user对象生成载荷（payload）ide
jwt_payload_handler(user) => payload
from rest_framework_jwt.serializers import jwt_payload_handler
（3）经过载荷签发token源码分析
jwt_encode_handler(payload) => token from rest_framework_jwt.serializers import jwt_encode_handler
（1）前提：一个配置JWT认证的视图类 就必须提供token进行认证 在认证的过程当中完成token的校验post
rest_framework_jwt.authentication.JSONWebTokenAuthentication 的 父类 BaseJSONWebTokenAuthentication 的 authenticate 方法
# 请求头拿认证信息jwt-token => 经过反爬小规则肯定有用的token => payload => user
（2）核心源码测试
rest_framework_jwt.authentication.BaseJSONWebTokenAuthentication的authenticate(self, request)方法
def authenticate(self, request): """ Returns a two-tuple of `User` and token if a valid signature has been supplied using JWT-based authentication. Otherwise returns `None`. """ # 带有反爬小规则的获取token：前台必须按 "jwt token字符串" 方式提交 # 校验user第1步：从请求头 HTTP_AUTHORIZATION 中拿token，并提取 jwt_value = self.get_jwt_value(request) # 游客 if jwt_value is None: return None # 校验 try: # 校验user第2步：token => payload payload = jwt_decode_handler(jwt_value) except jwt.ExpiredSignature: msg = _('Signature has expired.') raise exceptions.AuthenticationFailed(msg) except jwt.DecodeError: msg = _('Error decoding signature.') raise exceptions.AuthenticationFailed(msg) except jwt.InvalidTokenError: raise exceptions.AuthenticationFailed() # 校验user第3步：token => payload user = self.authenticate_credentials(payload) return (user, jwt_value)
（3）检验tokenurl
（1）从请求头中获取tokenspa
（2）根据token解析出payload（载荷）
jwt_decode_handler(token) => payloay from rest_framework_jwt.authentication import jwt_decode_handler
（3）根据载荷解析出用户
self.authenticate_credentials(payload) => user
继承drf-jwt的BaseJSONWebTokenAuthentication，拿到父级的authenticate_credentials方法
（1）model层
# 模型层 from django.db import models from django.contrib.auth.models import AbstractUser class User(AbstractUser): mobile = models.CharField(max_length=11, unique=True) class Meta: db_table = 'api_user' verbose_name = '用户表' verbose_name_plural = verbose_name def __str__(self): return self.username
（2）api/serializers序列化层
import re from rest_framework_jwt.serializers import jwt_payload_handler # 生成载荷 from rest_framework_jwt.serializers import jwt_encode_handler # 解析user from . import models # 导入模型层 from rest_framework import serializers # 序列化组件 class UserModelSerializer(serializers.ModelSerializer): # 定义校验类
# 自定义反序列字段：必定要设置write_only，只参与反序列化，不会与model类字段映射
user = serializers.CharField(write_only=True) pwd = serializers.CharField(write_only=True) class Meta: model = models.User fields = ['user', 'pwd', 'username', 'mobile', 'email'] extra_kwargs = { 'username': { 'read_only': True }, 'mobile': { 'read_only': True }, 'email': { 'read_only': True }, } def validate(self, attrs): # 设置校验的数据 user = attrs.get('user') pwd = attrs.get('pwd') # 设置校验的方式 if re.match(r'.*@.*', user): # 匹配邮箱号 进行邮箱登陆 user_obj = models.User.objects.filter(email=user).first() elif re.match(r'1[3-9][0-9]{9}', user): # 匹配13-19开头的手机号 进行手机登陆 user_obj = models.User.objects.filter(mobile=user).first() # 用户名登陆 else: user_obj = models.User.objects.filter(username=user).first() if user_obj and user_obj.check_password(pwd): # 判断是否有用户与传入的密码是否正确 payload = jwt_payload_handler(user_obj) # 生成载荷 print(payload) # {'user_id': 1, 'username': 'admin', 'exp': datetime.datetime(2019, 10, 23, 12, 37, 52, 240300), 'email': '1234@qq.com'} token = jwt_encode_handler(payload) # 生成token print(token) # eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxLCJ1c2VybmFtZSI6ImFkbWluIiwiZXhwIjoxNTcxODM0MjcyLCJlbWFpbCI6IjEyMzRAcXEuY29tIn0.zLeCBgT6j6zi2g59iTIGIB5ELSmiWmgf8IdMk1mZors self.user = user_obj # 将用户传给后端 self.token = token # 将token传入给后端 return attrs # 上述条件 都不知足 直接抛出异常 raise serializers.ValidationError( { 'msg': "数据错误" } )
'''
1:设置自定义反序列化字段
2：设置全局校验钩子
3：经过前台输入的数据 对数据进行匹配生成user_obj
4：将生成的user_obj 生成载荷
5：调用生成的载荷生成token
6：将生成user_obj 与 token传入后端
'''
（3）路由层
url(r'^login/$', views.Login.as_view()),
（4）视图层
from . import serializers class Login(APIView): permission_classes = [] # 不进行权限校验 authentication_classes = [] # 不进行认证校验 def post(self, request, *args, **kwargs): user_ser = serializers.UserModelSerializer(data=request.data) # 拿到传入的数据进行反序列化校验
user_ser.is_valid(raise_exception=True) # 序列化类校验获得登陆用户与token存放在序列化对象中 token = user_ser.token user_data = serializers.UserModelSerializer(user_ser.user).data # 获取传入的数据 进行序列化 return Response( { 'status': 0, 'msg': "测试成功", 'results': user_data, 'token': token } )
'''
1:获取前台传入的数据 进行反序列化 拿到user_ser
2:判断数据是否有效 若是无效直接抛出异常
3：在将前台传入的数据进行序列化 传入给前台展现
'''
（1）api/
import jwt from rest_framework_jwt.authentication import BaseJSONWebTokenAuthentication from rest_framework_jwt.authentication import jwt_decode_handler from rest_framework.exceptions import AuthenticationFailed class JWTAuthentication(BaseJSONWebTokenAuthentication): def authenticate(self, request): jwt_token = request.META.get('HTTP_AUTHORIZATION') # 自定义校验规则：auth token jwt token = self.parse_jwt_token(jwt_token) if token is None: return None try: # token => payload payload = jwt_decode_handler(token) except jwt.ExpiredSignature: raise AuthenticationFailed('token已过时') except: raise AuthenticationFailed('非法用户') # payload => user user = self.authenticate_credentials(payload) return (user, token) # 自定义校验规则：auth token jwt，auth为前盐，jwt为后盐 def parse_jwt_token(self, jwt_token): tokens = jwt_token.split() if len(tokens) != 3 or tokens[0].lower() != 'auth' or tokens[2].lower() != 'jwt': return None return tokens[1]
（2）视图层
from rest_framework.views import APIView from utils.response import APIResponse # 必须登陆后才能访问 - 经过了认证权限组件 from rest_framework.permissions import IsAuthenticated # 自定义jwt校验规则 from .authentications import JWTAuthentication class UserDetail(APIView): authentication_classes = [JWTAuthentication] permission_classes = [IsAuthenticated] def get(self, request, *args, **kwargs): return APIResponse(results={'username': request.user.username})
from django.contrib import admin from . import models # 自定义User表，admin后台管理，采用密文密码 from django.contrib.auth.admin import UserAdmin class MyUserAdmin(UserAdmin): add_fieldsets = ( (None, { 'classes': ('wide',), 'fields': ('username', 'password1', 'password2', 'mobile', 'email'), }), ) admin.site.register(models.User, MyUserAdmin)
（1）模型层
class Car(models.Model): name = models.CharField(max_length=16, unique=True, verbose_name='车名') price = models.DecimalField(max_digits=10, decimal_places=2, verbose_name='价格') brand = models.CharField(max_length=16, verbose_name='品牌') class Meta: db_table = 'api_car' verbose_name = '汽车表' verbose_name_plural = verbose_name def __str__(self): return self.name
（2）admin注册
admin.site.register(models.Car)
（3）序列化层api/
class CarModelSerializer(serializers.ModelSerializer): class Meta: model = models.Car fields = ['name', 'price', 'brand']
from rest_framework.generics import ListAPIView # 进行群查 from rest_framework.filters import SearchFilter # 导入搜索组件 from . import models class CarListAPIView(ListAPIView): queryset = models.Car.objects.all() serializer_class = serializers.CarModelSerializer filter_backends = [SearchFilter] # 局部过滤 search_fields = ['name','price'] # 设置查找的字段
from rest_framework.generics import ListAPIView # 第一步：drf的OrderingFilter - 排序过滤 from rest_framework.filters import OrderingFilter class CarListAPIView(ListAPIView): queryset = models.Car.objects.all() serializer_class = serializers.CarModelSerializer # 第二步：局部配置 过滤类 们（全局配置用DEFAULT_FILTER_BACKENDS） filter_backends = [OrderingFilter] # 第三步：OrderingFilter过滤类依赖的过滤条件 => 接口：/cars/?ordering=... ordering_fields = ['pk', 'price'] # eg：/cars/?ordering=-price,pk，先按price降序，若是出现price相同，再按pk升序
api/pahenations
from rest_framework.pagination import PageNumberPagination class MyPageNumberPagination(PageNumberPagination): # ?page=页码 page_query_param = 'page' # ?page=页面 下默认一页显示的条数 page_size = 3 # ?page=页面&page_size=条数 用户自定义一页显示的条数 page_size_query_param = 'page_size' # 用户自定义一页显示的条数最大限制：数值超过5也只显示5条 max_page_size = 5
视图层
from rest_framework.generics import ListAPIView class CarListAPIView(ListAPIView): # 若是queryset没有过滤条件，就必须 .all()，否则分页会出问题 queryset = models.Car.objects.all() serializer_class = serializers.CarModelSerializer # 分页组件 - 给视图类配置分页类便可 - 分页类须要自定义，继承drf提供的分页类便可 pagination_class = pagenations.MyPageNumberPagination
api/
from rest_framework.pagination import LimitOffsetPagination class MyLimitOffsetPagination(LimitOffsetPagination): # ?offset=从头偏移的条数&limit=要显示的条数 limit_query_param = 'limit' offset_query_param = 'offset' # ?不传offset和limit默认显示前3条，只设置offset就是从偏移位日后再显示3条 default_limit = 3 # ?limit能够自定义一页显示的最大条数 max_limit = 5 # 只使用limit结合ordering能够实现排行前几或后几 # ?ordering=-price&limit=2 => 价格前2
views
from rest_framework.generics import ListAPIView class CarListAPIView(ListAPIView): # 若是queryset没有过滤条件，就必须 .all()，否则分页会出问题 queryset = models.Car.objects.all() serializer_class = serializers.CarModelSerializer # 分页组件 - 给视图类配置分页类便可 - 分页类须要自定义，继承drf提供的分页类便可 pagination_class = pagenations.MyLimitOffsetPagination
api/
# 注：必须基于排序规则下进行分页 # 1）若是接口配置了OrderingFilter过滤器，那么url中必须传ordering # 1）若是接口没有配置OrderingFilter过滤器，必定要在分页类中声明ordering按某个字段进行默认排序 from rest_framework.pagination import CursorPagination class MyCursorPagination(CursorPagination): cursor_query_param = 'cursor' page_size = 3 page_size_query_param = 'page_size' max_page_size = 5 ordering = '-pk'
views
from rest_framework.generics import ListAPIView class CarListAPIView(ListAPIView): # 若是queryset没有过滤条件，就必须 .all()，否则分页会出问题 queryset = models.Car.objects.all() serializer_class = serializers.CarModelSerializer # 分页组件 - 给视图类配置分页类便可 - 分页类须要自定义，继承drf提供的分页类便可 pagination_class = pagenations.MyCursorPagination
api/filters
# 自定义过滤器，接口：?limit=显示的条数 class LimitFilter: def filter_queryset(self, request, queryset, view): # 前台固定用 ?limit=... 传递过滤参数 limit = request.query_params.get('limit') if limit: limit = int(limit) return queryset[:limit] return queryset
views
from rest_framework.generics import ListAPIView class CarListAPIView(ListAPIView): # 若是queryset没有过滤条件，就必须 .all()，否则分页会出问题 queryset = models.Car.objects.all() serializer_class = serializers.CarModelSerializer # 局部配置 过滤类 们（全局配置用DEFAULT_FILTER_BACKENDS） filter_backends = [LimitFilter]
（1）安装
pip3 install django-filter
（2）api/filter
# django-filter插件过滤器类 from django_filters.rest_framework.filterset import FilterSet from . import models # 自定义过滤字段 from django_filters import filters class CarFilterSet(FilterSet): min_price = filters.NumberFilter(field_name='price', lookup_expr='gte') max_price = filters.NumberFilter(field_name='price', lookup_expr='lte') class Meta: model = models.Car fields = ['brand', 'min_price', 'max_price'] # brand是model中存在的字段，通常都是能够用于分组的字段 # min_price、max_price是自定义字段，须要本身自定义过滤条件
（3）视图层
# django-filter插件过滤器 from django_filters.rest_framework import DjangoFilterBackend from .filters import CarFilterSet class CarListAPIView(ListAPIView): queryset = models.Car.objects.all() serializer_class = serializers.CarModelSerializer # 局部配置 过滤类 们（全局配置用DEFAULT_FILTER_BACKENDS） filter_backends = [DjangoFilterBackend] # django-filter过滤器插件使用 filter_class = CarFilterSet # 接口：?brand=...&min_price=...&max_price=... # eg:?brand=宝马&min_price=5&max_price=10 => 5~10间的宝马牌汽车