频率源码分析
频率源码
APIView----disaptch---》self.initial(request, *args, **kwargs)---》416行:self.check_throttles(request)----》352行 check_throttles def check_throttles(self, request): # self.get_throttles()就是咱们配置在视图类上频率类的对象列表[频率类对象,] for throttle in self.get_throttles(): # 执行频率类对象的allow_request,传了2个,返回True或False if not throttle.allow_request(request, self): # 反会给前端失败,显示还剩多长时间能再访问 throttle_durations.append(throttle.wait())
频率逻辑
1 写一个类,继承,BaseThrottle 2 在类中重写:allow_request方法,传入 3个参数 3 在allow_request写限制逻辑,如果还能访问--》返回True 4 如果超了次数,就不能访问,返回False 5 局部配置在视图类上 6 全局配置在配置文件中
SimpleRateThrottle 源码分析
SimpleRateThrottle内部一定有:allow_request---》 def allow_request(self, request, view): # 咱们没写,以后咱们可以在频率类中直接写 # rate='3/m' 以后不用写scope了,就会按一分钟访问3次现在 if self.rate is None: return True # 取出:重写的get_cache_key返回的值,咱们返回了访问者ip self.key = self.get_cache_key(request, view) if self.key is None: return True # 根据当前访问者ip,取出 这个人的访问时间列表 [访问时间1,访问2,访问3,访问4] self.history = self.cache.get(self.key, []) # 取出当前时间 self.now = self.timer() # 把访问时间列表中超过 限制时间外的时间剔除 while self.history and self.history[-1] <= self.now - self.duration: self.history.pop() # 判断访问时间列表是否大于 3 if len(self.history) >= self.num_requests: return self.throttle_failure() return self.throttle_success()
自定义频率类,实现一分钟只能访问三次的控制
(1)取出访问者ip (2)判断当前ip不在访问字典里,添加进去,并且直接返回True,表示第一次访问,在字典里,继续往下走 (3)循环判断当前ip的列表,有值,并且当前时间减去列表的最后一个时间大于60s,把这种数据pop掉,这样列表中只有60s以内的访问时间, (4)判断,当列表小于3,说明一分钟以内访问不足三次,把当前时间插入到列表第一个位置,返回True,顺利通过 (5)当大于等于3,说明一分钟内访问超过三次,返回False验证失败
我们在drf中写的时候,不需要继承 BaseThrottle,继承了SimpleRateThrottle,重写get_cache_key
自定义频率类
from rest_framework.throttling import BaseThrottle class MyThrottle(BaseThrottle): # VISIT_RECORD = {'192.168.1.1':[当前时间,当前时间,访问时间列表]} VISIT_RECORD = {} def __init__(self): self.history = [] def allow_request(self, request, view): ip = request.META.get('REMOTE_ADDR') import time ctime = time.time() # (2)判断当前ip不在访问字典里,说明是第一次访问,添加进去,并且直接返回True,表示第一次访问 if ip not in self.VISIT_RECORD: self.VISIT_RECORD[ip] = [ctime, ] # VISIT_RECORD = {'192.168.1.1':[当前时间2,当前时间1,]} return True # (3)循环判断当前ip的列表,有值,并且当前时间减去列表的最后一个时间大于60s,把这种数据pop掉,这样列表中只有60s以内的访问时间, self.history = self.VISIT_RECORD[ip] # 访问时间列表 while self.history and ctime - self.history[-1] > 60: # 循环删除1分钟之前访问的实际 self.history.pop() # 最后self.history都剩下是一分钟之内的时间了 # (4)判断,当列表小于3,说明一分钟以内访问不足三次,把当前时间插入到列表第一个位置,返回True,顺利通过 # (5)当大于等于3,说明一分钟内访问超过三次,返回False验证失败 if len(self.history) < 3: self.history.insert(0, ctime) return True else: return False def wait(self): import time ctime = time.time() return 60 - (ctime - self.history[-1])
过滤排序分页全局异常
过滤
#1 针对于查询所有接口---》继承:GenericAPIView+ListModelMixin---》只需要在视图类中写一个类属性---》filter_backends = [过滤类,过滤类2] # 过滤类: 1 内置的:SearchFilter 类属性:search_fields=[可以按字段过滤] 127.0.0.1:8080/books/?search=红 2 第三方 -django-filter -精准匹配 127.0.0.1:8080/books/?name=红楼梦 -能更强大 3 自定义的 写一个类,继承:BaseFilterBackend 重写 filter_queryset 在filter_queryset中完成过滤,会把qs传入,返回过滤后的qs即可 # 2 源码分析 ListModelMixin---》list方法---》queryset = self.filter_queryset(self.get_queryset()) 过滤后的数据----》执行了GenericAPIView的---》filter_queryset---》取出一个个配置在视图类上的过滤类,依次实例化得到对象后执行对象的filter_queryset方法完成过滤---》最终返回的数据,就是过滤后的数据 # 3 继承APIView写过滤 request 取出过滤条件 book_list=Book.objects.all().filter(过滤)
排序
1 针对于查询所有接口---》继承:GenericAPIView+ListModelMixin---》只需要在视图类中写一个类属性---》filter_backends = [排序类] 2 内置排序类即可:OrderingFilter---》配置类属性ordering_fields
分页
分页三种方式
三种分页方式,必须继承分页类,重写几个类属性实现--->配置在视图类上--》继承:GenericAPIView+ListModelMixin PageNumberPagination:用的最多,之前常见的分页方式,查询第几页,每页有多少条的分页方式 :page=10&size=3 LimitOffsetPagination:从第几条开始,取几条 offset=10&limit=3 从第10条开始,取3条 CursorPagination:只能上一页和下一页,需要排好序再分页
继承APIView 写分页
class Pager(APIView): def get(self,request,*args,**kwargs): # 获取所有数据 ret=models.Book.objects.all() # 创建分页对象 page=PageNumberPagination() # 在数据库中获取分页的数据 page_list=page.paginate_queryset(ret,request,view=self) # 对分页进行序列化 ser=BookSerializer1(instance=page_list,many=True) # return Response(ser.data) # 这个也是返回Response对象,但是比基本的多了上一页,下一页,和总数据条数(了解即可) return page.get_paginated_response(ser.data) return Response(ser.data) # 只会有数据,不会有上一页和下一页,总条数
全局异常
# 前后端分离了,后期,后端出了异常,我们不想让前端看到,我们需要捕获全局异常,统一返回格式 # drf 源码中已经处理 APIView--->dispatch---> try: # 1 执行三大认证 # 2 执行视图类的方法 except Exception as exc: response = self.handle_exception(exc) -463行左右: # exception_handler就是配置文件中配置的一个函数-->默认的 # 后期自己写了 exception_handler = self.get_exception_handler() response = exception_handler(exc, context) # 默认执行的是:rest_framework.views.exception_handler函数---》只处理了drf的异常 # 咱们处理全局异常的步骤: 1 写一个函数,在内部处理 from rest_framework.views import exception_handler def common_exception_handler(exc,context): res=exception_handler(exc,context) if res: #这次异常是drf的,并且它处理了 # 我们要统一返回格式 return Response({'code':888,'msg':"系统异常(drf异常),请联系系统管理员:%s"%res.data.get('detail',None)}) else: return Response({'code': 999, 'msg': "系统异常(非drf异常),请联系系统管理员:%s" % str(exc)}) 2 配置在配置文件上
接口文档
# 接口写完,必须编写接口文档,给别人用 # 接口文档规范:要有什么东西 1 描述 2 地址 3 请求方式 4 请求编码格式 5 请求参数,详解 6 返回示例 json 7 返回示例中字段详解 # 编写的位置: 1 直接写在文件中共享(word,md) 2 平台上写 - 公司搭建的平台(自己研发,第三方开源) - 使用第三方接口文档平台 3 自动生成 -coreaip
JWT认证
JWT认证
json web tokne缩写,一种前后端认证的方法,区别与session的方案,不需要在后端存储数据,也能实现会话保持
JWT原理
三段式: 签发阶段 登录: 用户名+密码 手机号+验证码 本机号码一键登录--》向后端就只传了手机号--》根据手机号签发token 认证阶段 drf中的认证类 校验token是否合法
别人截获到token后,模拟发送请求
解决方法
1 设置过期时间---》10m 2 需要登录后才能访问的接口,不仅要带token还是带个加密串 sign 3 双token认证 返回两个token:一个token:7天过期,一个token:10分钟过期 以后用就只带10分钟过期的token 过了10分钟了,10分钟token失效了 携带着7的token到后端,生成一个10分钟的返回 原来的token串使用md5+盐生成一个串--》签名
自定义用户表
多方式登录
权限控制
acl:访问控制列表
用户跟权限多对多
基于角色的访问控制 rbac
用户跟角色关系,角色和权限管理 用户表 角色表 权限表 用户和角色多对多中间表 角色和权限多对多中间表 用户和权限多对多
abac:
基于属性+角色的访问控制
基于属性+访问控制列表 张三:【看视频,发视频】 张三:晚上12点凌晨7点 不能发视频
标签:---,return,self,request,列表,day14,访问,drf From: https://www.cnblogs.com/shanghaipudong/p/17702607.html