1.不存数据库的token认证
import hashlib from day98 import settings def check_token(token): ret = True user_info = None try: ll = token.split('|') # da89744b701b5d8bc5b9a76b4ddb3dd4 , {"name": "cao", "id": 1}，已经切分成了一个列表 md5 = hashlib.md5() # 需要给这个{"name": "cao", "id": 1}加密，它就是列表里的第一个值 md5.update(ll[1].encode('utf-8')) # 在setting里面全局配置一下，给token加盐 md5.update(settings.password.encode('utf-8')) # hex=da89744b701b5d8bc5b9a76b4ddb3dd4 hex = md5.hexdigest() if not hex == ll[0]: ret = False else: user_info = ll[1] except Exception as e: ret = False return ret, user_info class LoginAuth(BaseAuthentication): # 函数名一定要叫authenticate，需要接收2个参数，第二个参数是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: # 能查到，说明认证通过，反回空 # ret.user就是当前登录用户对象 return user_info, None # 如果查不到，就抛出异常 raise exceptions.APIException('认证失败')
class Books(APIView): # 列表中类名不能加括号 authentication_classes = [LoginAuth, ] def get(self, request, *args, **kwargs): # 只要通过认证，就能取到当前登录用户对象的密码，id等信息 # print(request.query_params) print(request.user) # print(request.user.pwd) response = {'status': 100, 'msg': '查询成功'} res = models.Book.objects.all() book_ser = MySerializer.BookSerializer(res, many=True) # 这个数据是需要返回给前台的 response['data'] = book_ser.data # print(book_ser.data) return JsonResponse(response, safe=False) # 登录接口,不存数据库的token认证 import json from day98 import settings def create_token(user_pk): md5 = hashlib.md5() md5.update(user_pk.encode('utf-8')) # 在setting里面全局配置一下，给token加盐 md5.update(settings.password.encode('utf-8')) hex = md5.hexdigest() token = '|'.join([hex, user_pk]) # token=hex+'|'+user_info print(token) return token class Login(APIView): authentication_classes = [] # 登录就是使用post，get是返回一个页面 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}) # 生成vfvevberber|{'name': user.name, 'id': user.pk}的token token = create_token(str(user.pk)) # 登陆成功之后把登录返回给他，以后就带着token过来 response['token'] = token except ObjectDoesNotExist as e: response['status'] = 101 response['msg'] = '用户名或密码错误' except Exception as e: # 万能异常，里面只要出错，程序就会走到这里 response['status'] = 102 # response['msg'] = '未知错误' # 把整个错误信息转换成str类型，赋值给e，一般在测试时使用这个 response['msg'] = str(e) # 如果不写safe=False，只能序列化字典形式，如果字典里面又套了列表，或者直接是一个列表，就必须写safe=False return JsonResponse(response, safe=False)
频率简介：
节流，访问控制。为了控制用户对某个url请求的频率，比如，一分钟以内，只能访问三次
内置的访问频率控制类：SimpleRateThrottle
-写一个类,继承SimpleRateThrottle
2.自定义频率类，自定义频率规则
from rest_framework.throttling import BaseThrottle import time
MyAuth.py-频率控制 # 频率控制
'''
自定义的逻辑：
{'ip1':[时间1 ,时间2],
'ip2':[时间1, ],
（1）取出访问者ip
（2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走
（3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间
（4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
（5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败
'''
class MyThrottle():
visitor_dic = {}
def __init__(self):
self.history=None
def allow_request(self, request, view):
# META:请求所有的东西的字典
# 拿出ip地址
ip = request.META.get('REMOTE_ADDR')
# 判断id在不在字典里面，不在字典里面，说明是第一次访问
ctime = time.time()
if ip not in self.visitor_dic:
self.visitor_dic[ip] = [ctime, ]
return True
# 根据当前访问者的ip，取出访问者的时间列表
history = self.visitor_dic[ip]
# 记录一下当前访问者的ip
self.history=history
# 如果时间大于这个值，说明是一分钟之后访问的，那么就删除一分钟之后访问的
while history and ctime - history[-1] > 60:
history.pop()
# 一分钟访问小于三次
if len(history) < 3:
# 就把当前时间放在第0个位置上
history.insert(0, ctime)
return True
else:
return False
def wait(self):
# 剩余时间 # 要取列表里的最后一个值
ctime=time.time() # 当前时间减去最后一次访问的时间
return 60-(ctime-self.history[-1])
view视图层
from django.shortcuts import render, HttpResponse
from rest_framework import exceptions
from rest_framework.views import APIView
from app01.myAuth import MyThrottle
from rest_framework.parsers import JSONParser, FormParser
class Test(APIView):
throttle_classes = [MyThrottle, ]
def get(self, request):
return HttpResponse('ok')
3.内置频率类及局部使用
-内置的访问频率控制类（根据ip限制）：SimpleRateThrottle -写一个类,继承SimpleRateThrottle MyAuth.py-频率控制 from rest_framework.throttling import BaseThrottle, SimpleRateThrottle class MyThrottle(SimpleRateThrottle): scope = 'aaa' def get_cache_key(self, request, view): # 返回ip地址 # ip=request.META.get('REMOTE_ADDR') # return ip return self.get_ident(request)
settings.py
STATIC_URL = '/static/'
REST_FRAMEWORK = {
# 'DEFAULT_THROTTLE_CLASSES': ['app01.myAuth.MyThrottle', ],
'DEFAULT_THROTTLE_RATES': {
# 每分钟访问10次
'aaa': '5/m'
}
}
使用： 局部使用: -在视图类中写 throttle_classes = [MyThrottle,]
全局使用: 在settings中配置：
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': ['app01.MyAuth.LoginAuth', ],
# 'DEFAULT_PERMISSION_CLASSES': ['app01.MyAuth.UserPermission', ]
}
-局部禁用:
-在视图类中写
throttle_classes = []
4.错误信息改成中文显示
def throttled(self, request, wait): class MyThrottled(exceptions.Throttled): default_detail = '傻逼' extra_detail_singular = '还剩 {wait} 秒.' extra_detail_plural = '还剩 {wait} 秒' raise MyThrottled(wait)