一、不存数据库的认证

环境：

与上次（Rest Framework：四）数据库连接的一致，数据也一致，这次也使用的是上次的django环境

urls.py

url(r'^login/', views.Login.as_view()),
url(r'^books/', views.Books.as_view()),

models.py

from django.db import models# Create your models here.
# 用户信息
class UserInfo(models.Model):name = models.CharField(max_length=32)# 写choiceuser_choice = ((0, '普通用户'), (1, '会员'), (2, '超级用户'))# 指定choice,可以快速的通过数字,取出文字user_type = models.IntegerField(choices=user_choice, default=0)pwd = models.CharField(max_length=32)# 用户token
class UserToken(models.Model):token = models.CharField(max_length=64)user = models.OneToOneField(to=UserInfo)class Book(models.Model):nid = models.AutoField(primary_key=True)name = models.CharField(max_length=32)price = models.DecimalField(max_digits=5, decimal_places=2)publish_date = models.DateField()publish = models.ForeignKey(to='Publish', to_field='nid', on_delete=models.CASCADE)authors = models.ManyToManyField(to='Author')def __str__(self):return self.nameclass Author(models.Model):nid = models.AutoField(primary_key=True)name = models.CharField(max_length=32)age = models.IntegerField()author_detail = models.OneToOneField(to='AuthorDatail', to_field='nid', unique=True, on_delete=models.CASCADE)class AuthorDatail(models.Model):nid = models.AutoField(primary_key=True)telephone = models.BigIntegerField()birthday = models.DateField()addr = models.CharField(max_length=64)class Publish(models.Model):nid = models.AutoField(primary_key=True)name = models.CharField(max_length=32)city = models.CharField(max_length=32)email = models.EmailField()def __str__(self):return self.namedef test(self):return self.email

迁移数据库

python3 manage makemigrations
python3 manage migrate

app01/MySerializer.py

from rest_framework import serializers
from app01 import modelsclass BookSerializer(serializers.ModelSerializer):class Meta:model = models.Bookfields = '__all__'class AuthorSerializer(serializers.ModelSerializer):class Meta:model = models.Authorfields = '__all__'class UserSer(serializers.ModelSerializer):class Meta:model = models.UserInfofields='__all__'# 显示用户类型以中文显示user_type=serializers.CharField(source='get_user_type_display')# 上面一句话就ok# user_type=serializers.SerializerMethodField()# def get_user_type(self,obj):#     return obj.get_user_type_display()

app01/MyAuth.py

from rest_framework import exceptions
from app01 import models
from rest_framework.authentication import BaseAuthenticationimport hashlib
from pro_12_17 import settingsdef check_token(token):ret = Trueuser_info=Nonetry:li = token.split('|')md5 = hashlib.md5()md5.update(li[1].encode('utf-8'))md5.update(settings.password.encode('utf-8'))hex = md5.hexdigest()if not hex == li[0]:ret = Falseelse:user_info=li[1]except Exception as e:ret = Falsereturn ret,user_info# 用dnf 认证，写一个类
class LoginAuth(BaseAuthentication):# 函数名必须叫这个名字，接收必须两个参数，第二个参数是request对像def authenticate(self, request):# 从request对像中取出token （也可以从其他地方中取）token = request.query_params.get('token')# ret 是布尔类型，表示验证通过或失败，user_info是user的字典ret, user_info = check_token(token)if ret:# 可以查到，说明认证通过，返回空return user_info, None#   否则会报异常raise exceptions.APIException('认证失败')from rest_framework.permissions import BasePermissionclass UserPermission(BasePermission):# message是出错显示的中文message = '没权限查看'def has_permission(self, request, view):print(request.user)user_type = request.user.user_typeprint(user_type)# 取出用户类型对应的文字# 固定用法:get_字段名字_display()user_type_name = request.user.get_user_type_display()print(user_type_name)if user_type == 2:return Trueelse:return False

views.py

from django.shortcuts import render
from django.http import JsonResponse
from rest_framework.views import APIView
from app01 import models
import hashlib
import time
from django.core.exceptions import ObjectDoesNotExist
from app01 import MySerializer# Create your views here.def get_token(name):# 生成一个md5对象md5 = hashlib.md5()# 往里添加值,必须是bytes格式# time.time()生成时间戳类型,转成字符串,再encode转成bytes格式md5.update(str(time.time()).encode('utf-8'))md5.update(name.encode('utf-8'))return md5.hexdigest()from rest_framework import exceptions
from app01.MyAuth import LoginAuth
from rest_framework.request import Requestclass Books(APIView):# 列表中类型不能加括号# 认证组件局部使用# authentication_classes = [LoginAuth, ]# 认证组件，局部禁用 authentication_classes = []authentication_classes = []def get(self, request, *args, **kwargs):print(request.user)response = {'status': 100, 'msg': '查询成功'}ret = models.Book.objects.all()book_ser = MySerializer.BookSerializer(ret, many=True)response['data'] = book_ser.datareturn JsonResponse(response, safe=False)# 只能超级用户可以查看作者详情，其他人不能看
from app01.MyAuth import UserPermissionimport json
from pro_12_17 import settingsdef create_token(user_id):md5 = hashlib.md5()md5.update(user_id.encode('utf-8'))md5.update(settings.password.encode('utf-8'))hex = md5.hexdigest()token = hex + '|' + user_idprint(token)return tokenclass Login(APIView):authentication_classes = []def post(self, request, *args, **kwargs):response = {'status': 100, 'msg': '登录成功'}name = request.data.get('name')pwd = request.data.get('pwd')try:user = models.UserInfo.objects.get(name=name, pwd=pwd)user_info_json = json.dumps({"name": user.name, "id": user.pk})token = create_token(str(user.pk))response['token'] = tokenexcept ObjectDoesNotExist as e:response['status'] = 101response['msg'] = '用户名或密码错误'except Exception as e:response['status'] = 102# response['msg']='未知错误'response['msg'] = str(e)return JsonResponse(response, safe=False)

settings.py

REST_FRAMEWORK={'DEFAULT_AUTHENTICATION_CLASSES':['app01.MyAuth.LoginAuth',],# 'DEFAULT_PERMISSION_CLASSES':['app01.MyAuth.UserPermission',],
}
password='123'

使用postman验证结果：

先登录：

后查询结果：

二、自定义的频率控制

新环境：

urls.py

url(r'^test/', views.Test.as_view()),

views.py

from django.shortcuts import render,HttpResponse# Create your views here.
from rest_framework.views import APIView
from app01.MyAuth import MyThrotte
class Test(APIView):throttle_classes = [MyThrotte,]def get(self,request):return HttpResponse('ok')

app01/MyAuth.py

from rest_framework.throttling import BaseThrottle, SimpleRateThrottle
import time'''
#             {'ip1':[时间1 ,时间2],
#             'ip2':[时间1, ],
#             }
#             # （1）取出访问者ip
#             # （2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走
#             # （3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间，
#             # （4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
#             # （5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败
#         '''class MyThrotte():visitor_dic = {}def __init__(self):self.history = Nonedef allow_request(self, request, view):# Meta:请求所有的东西的字典# 拿出ip地址ip = request.META.get('REMOTE_ADDR')# 不在字典中,说明是第一次访问ctime = time.time()if ip not in self.visitor_dic:self.visitor_dic[ip] = [ctime, ]return True# 根据当前访问者ip,取出访问的时间列表history = self.visitor_dic[ip]self.history=historywhile history and ctime - history[-1] > 60:history.pop()if len(history) < 3:# 把当前时间放到第0个位置上history.insert(0, ctime)return Truereturn Falsedef wait(self):# 剩余时间ctime = time.time()return 60 - (ctime - self.history[-1])

测试访问：

get 请求方式访问：

http://127.0.0.1:8000/test/，多刷新几次。

三、内置频率控制类的使用（比上面的一种更为方便，简单）

urls.py

url(r'^test/', views.Test.as_view()),

views.py

from django.shortcuts import render,HttpResponse# Create your views here.
from rest_framework.views import APIView
from app01.MyAuth import MyThrottle
class Test(APIView):# 局部使用，打开全局，把局部要关闭掉# throttle_classes = [MyThrottle,]def get(self,request):return HttpResponse('ok')# 显示返回用户信息为中文字符def throttled(self, request, wait):class MyThrottled(exceptions.Throttled):default_detail = '剩余时间'extra_detail_singular = '还剩 {wait} 秒.'extra_detail_plural = '还剩 {wait} 秒'raise MyThrottled(wait)

app01/MyAuth.py

from rest_framework.throttling import BaseThrottle, SimpleRateThrottle
import timeclass MyThrottle(SimpleRateThrottle):scope = 'aaa'   # scope 是内置的，必须叫这个名名def get_cache_key(self, request, view):# 返回ip地址# ip=request.META.get('REMOTE_ADDR')# return ip# 下面这种比上面的更方便return self.get_ident(request)

settings.py

REST_FRAMEWORK = {# 全局使用：'DEFAULT_THROTTLE_CLASSES''DEFAULT_THROTTLE_CLASSES': ['app01.MyAuth.MyThrottle', ],'DEFAULT_THROTTLE_RATES': {'aaa': '10/m'     # 一分钟可以访问10次}
}

最后测试：

使用postman 测试：

四、解析器

作用：传过来的数据,解析成字典

settings.py

# 全局解析器
REST_FRAMEWORK = {'DEFAULT_PARSER_CLASSES':['rest_framework.parsers.JSONParser']}

urls.py

url(r'^test/', views.Test.as_view()),

app01/MyAuth.py

from rest_framework.throttling import BaseThrottle, SimpleRateThrottle
class MyThrottle(SimpleRateThrottle):scope = 'aaa'   # scope 是内置的，必须叫这个名名def get_cache_key(self, request, view):# 返回ip地址# ip=request.META.get('REMOTE_ADDR')# return ip# 下面这种比上面的更方便return self.get_ident(request)

views.py

from django.shortcuts import render,HttpResponse
from rest_framework import exceptions
from rest_framework.views import APIView
from rest_framework.parsers import JSONParser,FormParser
class Test(APIView):# 局部使用，打开全局，把局部要关闭掉# throttle_classes = [MyThrottle,]# 默认解析的格式有：json,form,multi 三种格式，不写默认支持三种parser_classes = [JSONParser,FormParser]   # 最好只写上你在使用的就好def get(self,request):return HttpResponse('ok')def post(self, request):print(request.data)return HttpResponse('post')# 显示返回用户信息为中文字符def throttled(self, request, wait):class MyThrottled(exceptions.Throttled):default_detail = '剩余时间'extra_detail_singular = '还剩 {wait} 秒.'extra_detail_plural = '还剩 {wait} 秒'raise MyThrottled(wait)

使用postman 测试功能：

成功。

不成功时是这样的：