rest-framework------频率控制
一: 频率简介
为了控制用户对某个url请求的频率,比如,一分钟只能访问5次
二:自定义频率类,自定义频率规则
自定义的逻辑: 1 取出访问者的ip 2 判断当前ip不在访问字典里,添加进去,并且直接返回True,表示第一次访问,在字典里,继续往下走 3 循环判断当前ip的列表,有值,并且当前时间减去列表的最后一个时间大于60s,把这种数据pop掉,,这样列表里面就是有60s以内的访问时间 4 判断,当列表小于3,说明一分钟以内访问不足三次,把当前时间插入到列表第一个位置,返回True,顺利通过 5 当大于3等于3,说明一分钟访问超过三次,返回False验证失败
自定义频率代码
class MyThrottle(): visitor_dic = {} def __init__(self): self.history = None def allow_request(self, request, view): ''' {'ip1':[时间1 ,时间2], 'ip2':[时间1, ], } #（1）取出访问者ip # （2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走 # （3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间， # （4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过 # （5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败 ''' # 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 = history while history and ctime - history[-1] > 60: history.pop() if len(history) < 3: # 把当前时间放到第0个位置上 history.insert(0, ctime) return True return False def wait(self): # 剩余时间 ctime = time.time() return 60 - (ctime - self.history[-1])
三:内置频率类及局部使用
写一个类,继承自SimpleRateThrottle,(根据ip限制)
from rest_framework.throttling import SimpleRateThrottle
class Mythrottle(SimpleRateThrottle): scope='abc' def get_cache_key(self, request, view): #返回ip地址 return self.get_ident(request)
在setting里配置:(一分钟访问三次)
REST_FRAMEWORK = {'DEFAULT_THROTTLE_RATES': { 'abc': '10/m' } }
ps:3/m指的是一分钟访问3次,m可以换成(s/h/d)代表的是秒/小时/天
Period should be one of: ('s', 'sec', 'm', 'min', 'h', 'hour', 'd', 'day')
在视图类里面用
throttle_classes = [MyThrottle,]
全局使用,把视图类里面的注掉,在settings中配置
'DEFAULT_THROTTLE_CLASSES': ['app01.MyAuth.MyThrottle', ], 'DEFAULT_THROTTLE_RATES': { 'aaa': '10/m' },
错误信息的中文显示,在视图类中
from django.shortcuts import render, HttpResponse from rest_framework import exceptions from rest_framework.views import APIView from app01.MyAuth import Mythrottle class Test(APIView): throttle_classes = [Mythrottle, ] # 局部使用 def get(self, request, *args, **kwargs): return HttpResponse("get") # --------下面是错误是显示中文的------- def throttled(self, request, wait): class MyThrottle(exceptions.Throttled): default_detail = _('小兔崽子') extra_detail_singular = '还剩 {wait} 秒.' extra_detail_plural = '还剩 {wait} 秒' raise MyThrottle(wait)
内置频率限制类：
BaseThrottle是所有类的基类：方法：def get_ident(self, request)获取标识，其实就是获取ip，自定义的需要继承它
AnonRateThrottle：未登录用户ip限制，需要配合auth模块用
SimpleRateThrottle：重写此方法，可以实现频率现在，不需要咱们手写上面自定义的逻辑
UserRateThrottle:登录用户频率限制，这个得配合auth模块来用
ScopedRateThrottle：应用在局部视图上的（忽略）
四:内置频率及全局使用
REST_FRAMEWORK = { 'DEFAULT_THROTTLE_CLASSES':['app01.utils.VisitThrottle',], 'DEFAULT_THROTTLE_RATES':{ 'luffy':'3/m' } }
五:源码分析
def check_throttles(self, request): for throttle in self.get_throttles(): if not throttle.allow_request(request, self): self.throttled(request, throttle.wait()) def throttled(self, request, wait): #抛异常，可以自定义异常，实现错误信息的中文显示 raise exceptions.Throttled(wait)
class SimpleRateThrottle(BaseThrottle): # 咱自己写的放在了全局变量，他的在django的缓存中 cache = default_cache # 获取当前时间，跟咱写的一样 timer = time.time # 做了一个字符串格式化， cache_format = 'throttle_%(scope)s_%(ident)s' scope = None # 从配置文件中取DEFAULT_THROTTLE_RATES，所以咱配置文件中应该配置，否则报错 THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES def __init__(self): if not getattr(self, 'rate', None): # 从配置文件中找出scope配置的名字对应的值，比如咱写的‘3/m’,他取出来 self.rate = self.get_rate() # 解析'3/m',解析成 3 m self.num_requests, self.duration = self.parse_rate(self.rate) # 这个方法需要重写 def get_cache_key(self, request, view): """ Should return a unique cache-key which can be used for throttling. Must be overridden. May return `None` if the request should not be throttled. """ raise NotImplementedError('.get_cache_key() must be overridden') def get_rate(self): if not getattr(self, 'scope', None): msg = ("You must set either `.scope` or `.rate` for '%s' throttle" % self.__class__.__name__) raise ImproperlyConfigured(msg) try: # 获取在setting里配置的字典中的之，self.scope是 咱写的luffy return self.THROTTLE_RATES[self.scope] except KeyError: msg = "No default throttle rate set for '%s' scope" % self.scope raise ImproperlyConfigured(msg) # 解析 3/m这种传参 def parse_rate(self, rate): """ Given the request rate string, return a two tuple of: <allowed number of requests>, <period of time in seconds> """ if rate is None: return (None, None) num, period = rate.split('/') num_requests = int(num) # 只取了第一位，也就是 3/mimmmmmmm也是代表一分钟 duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]] return (num_requests, duration) # 逻辑跟咱自定义的相同 def allow_request(self, request, view): """ Implement the check to see if the request should be throttled. On success calls `throttle_success`. On failure calls `throttle_failure`. """ if self.rate is None: return True self.key = self.get_cache_key(request, view) if self.key is None: return True self.history = self.cache.get(self.key, []) self.now = self.timer() # Drop any requests from the history which have now passed the # throttle duration while self.history and self.history[-1] <= self.now - self.duration: self.history.pop() if len(self.history) >= self.num_requests: return self.throttle_failure() return self.throttle_success() # 成功返回true，并且插入到缓存中 def throttle_success(self): """ Inserts the current request's timestamp along with the key into the cache. """ self.history.insert(0, self.now) self.cache.set(self.key, self.history, self.duration) return True # 失败返回false def throttle_failure(self): """ Called when a request to the API has failed due to throttling. """ return False def wait(self): """ Returns the recommended next request time in seconds. """ if self.history: remaining_duration = self.duration - (self.now - self.history[-1]) else: remaining_duration = self.duration available_requests = self.num_requests - len(self.history) + 1 if available_requests <= 0: return None return remaining_duration / float(available_requests) SimpleRateThrottle源码分析
rest-framework-------解析器
一: 解析器的作用
根据请求头content-type选择对应的解析器对请求体内容进行处理 有application/json,form-data等格式
概念:把前台传过来的数据,解析成能处理的数据类型(字典类型)
二:全局使用解析器
在settings中
REST_FRAMEWORK = { 'DEFAULT_PARSER_CLASSES':[ 'rest_framework.parsers.JSONParser' 'rest_framework.parsers.FormParser' 'rest_framework.parsers.MultiPartParser' ] }
路由:
urlpatterns = [ url(r'test/', TestView.as_view()), ]
视图函数
from rest_framework.views import APIView from rest_framework.response import Response class Test(APIView): def post(self,request,*args,**kwargs): print(request,content_type) print(request.data) print(request.POST) print(request.FILES) return Response('POST请求,相应内容') def put(self,request,*args,**kwargs): return Response('put请求,响应内容)
三:局部使用解析器
1. 仅处理请求头content-type为application/json的请求体
from django.conf.urls import url, include from web.views.s5_parser import TestView urlpatterns = [ url(r'test/', TestView.as_view(), name='test'), ]
#!/usr/bin/env python # -*- coding:utf-8 -*- from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.request import Request from rest_framework.parsers import JSONParser class TestView(APIView):
#局部使用解析器如下 parser_classes = [JSONParser, ] def post(self, request, *args, **kwargs): print(request.content_type) # 获取请求的值，并使用对应的JSONParser进行处理 print(request.data) # application/x-www-form-urlencoded 或 multipart/form-data时，request.POST中才有值 print(request.POST) print(request.FILES) return Response('POST请求，响应内容') def put(self, request, *args, **kwargs): return Response('PUT请求，响应内容')
2. 仅处理请求头content-type为application/x-www-form-urlencoded 的请求体
from django.conf.urls import url, include from web.views import TestView urlpatterns = [ url(r'test/', TestView.as_view(), name='test'), ]
#!/usr/bin/env python # -*- coding:utf-8 -*- from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.request import Request from rest_framework.parsers import FormParser class TestView(APIView): parser_classes = [FormParser, ] def post(self, request, *args, **kwargs): print(request.content_type) # 获取请求的值，并使用对应的JSONParser进行处理 print(request.data) # application/x-www-form-urlencoded 或 multipart/form-data时，request.POST中才有值 print(request.POST) print(request.FILES) return Response('POST请求，响应内容') def put(self, request, *args, **kwargs): return Response('PUT请求，响应内容')
3. 仅处理请求头content-type为multipart/form-data的请求体
from django.conf.urls import url, include from web.views import TestView urlpatterns = [ url(r'test/', TestView.as_view(), name='test'), ]
#!/usr/bin/env python # -*- coding:utf-8 -*- from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.request import Request from rest_framework.parsers import MultiPartParser class TestView(APIView): parser_classes = [MultiPartParser, ] def post(self, request, *args, **kwargs): print(request.content_type) # 获取请求的值，并使用对应的JSONParser进行处理 print(request.data) # application/x-www-form-urlencoded 或 multipart/form-data时，request.POST中才有值 print(request.POST) print(request.FILES) return Response('POST请求，响应内容') def put(self, request, *args, **kwargs): return Response('PUT请求，响应内容')
4. 仅上传文件
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Title</title> </head> <body> <form action="http://127.0.0.1:8000/test/" method="post" enctype="multipart/form-data"> <input type="text" name="user" /> <input type="file" name="img"> <input type="submit" value="提交"> </form> </body> </html>
#!/usr/bin/env python # -*- coding:utf-8 -*- from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.request import Request from rest_framework.parsers import FileUploadParser class TestView(APIView): parser_classes = [FileUploadParser, ] def post(self, request, filename, *args, **kwargs): print(filename) print(request.content_type) # 获取请求的值，并使用对应的JSONParser进行处理 print(request.data) # application/x-www-form-urlencoded 或 multipart/form-data时，request.POST中才有值 print(request.POST) print(request.FILES) return Response('POST请求，响应内容') def put(self, request, *args, **kwargs): return Response('PUT请求，响应内容')
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Title</title> </head> <body> <form action="http://127.0.0.1:8000/test/f1.numbers" method="post" enctype="multipart/form-data"> <input type="text" name="user" /> <input type="file" name="img"> <input type="submit" value="提交"> </form> </body> </html>
5. 同时多个Parser
from django.conf.urls import url, include from web.views import TestView urlpatterns = [ url(r'test/', TestView.as_view(), name='test'), ]
#!/usr/bin/env python # -*- coding:utf-8 -*- from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.request import Request from rest_framework.parsers import JSONParser, FormParser, MultiPartParser class TestView(APIView): parser_classes = [JSONParser, FormParser, MultiPartParser, ] def post(self, request, *args, **kwargs): print(request.content_type) # 获取请求的值，并使用对应的JSONParser进行处理 print(request.data) # application/x-www-form-urlencoded 或 multipart/form-data时，request.POST中才有值 print(request.POST) print(request.FILES) return Response('POST请求，响应内容') def put(self, request, *args, **kwargs): return Response('PUT请求，响应内容')
当同时使用多个parser时，rest framework会根据请求头content-type自动进行比对，并使用对应parser
四:源码分析
1 在调用request.data时，才进行解析，由此入手
@property
def data(self):
if not _hasattr(self, '_full_data'):
self._load_data_and_files()
return self._full_data
2 查看self._load_data_and_files()方法---->self._data, self._files = self._parse()
def _parse(self):
#用户请求头里content_type的值
media_type = self.content_type
#self.parsers 就是用户配置的parser_classes = [FileUploadParser,FormParser ]
#self里就有content_type，传入此函数
parser = self.negotiator.select_parser(self, self.parsers)
3 查看self.negotiator.select_parser(self, self.parsers)
def select_parser(self, request, parsers):
#同过media_type和request.content_type比较，来返回解析器，然后调用解析器的解析方法
#每个解析器都有media_type = 'multipart/form-data'属性
for parser in parsers:
if media_type_matches(parser.media_type, request.content_type):
return parser
return None
4 最终调用parser的解析方法来解析parsed = parser.parse(stream, media_type, self.parser_context)
1 Request实例化，parsers=self.get_parsers()
Request(
request,
parsers=self.get_parsers(),
authenticators=self.get_authenticators(),
negotiator=self.get_content_negotiator(),
parser_context=parser_context
)
2 get_parsers方法，循环实例化出self.parser_classes中类对象
def get_parsers(self):
return [parser() for parser in self.parser_classes]
3 self.parser_classes 先从类本身找，找不到去父类找即APIVIew 中的
parser_classes = api_settings.DEFAULT_PARSER_CLASSES
4 api_settings是一个对象，对象里找DEFAULT_PARSER_CLASSES属性，找不到，会到getattr方法
def __getattr__(self, attr):
if attr not in self.defaults:
raise AttributeError("Invalid API setting: '%s'" % attr)
try:
#调用self.user_settings方法，返回一个字典，字典再取attr属性
val = self.user_settings[attr]
except KeyError:
# Fall back to defaults
val = self.defaults[attr]
# Coerce import strings into classes
if attr in self.import_strings:
val = perform_import(val, attr)
# Cache the result
self._cached_attrs.add(attr)
setattr(self, attr, val)
return val
5 user_settings方法 ，通过反射去setting配置文件里找REST_FRAMEWORK属性，找不到，返回空字典
@property
def user_settings(self):
if not hasattr(self, '_user_settings'):
self._user_settings = getattr(settings, 'REST_FRAMEWORK', {})
return self._user_settings
频率组件: -频率是什么?节流,访问控制 -内置的访问频率控制类SimpleRateThrottle -写一个类,继承SimpleRateThrottle -class MyThrottle(SimpleRateThrottle): scope='aaa' def get_cache_key(self, request, view): return self.get_ident(request) -在setting中: REST_FRAMEWORK = { 'DEFAULT_THROTTLE_RATES':{ 'aaa':'10/m' } } -使用 局部使用: -在视图类中写 throttle_classes = [MyThrottle,] 全局使用: -在setting中写 'DEFAULT_THROTTLE_CLASSES':['app01.MyAuth.MyThrottle',], -局部禁用: -在视图类中写 throttle_classes = [] 错误信息改成中文显示: def throttled(self, request, wait): class MyThrottled(exceptions.Throttled): default_detail = '傻逼' extra_detail_singular = '还剩 {wait} 秒.' extra_detail_plural = '还剩 {wait} 秒' raise MyThrottled(wait) 解析器: 作用:传过来的数据,解析成字典 使用: 局部使用: from rest_framework.parsers import JSONParser,FormParser 在视图类中: parser_classes = [FormParser,] 全局使用 REST_FRAMEWORK = { 'DEFAULT_PARSER_CLASSES':[ 'rest_framework.parsers.JSONParser' ] } 局部使用指定的解析器: 在视图类中: parser_classes = [FormParser,]
Period should be one of: ('s', 'sec', 'm', 'min', 'h', 'hour', 'd', 'day')