目錄html
1、認證python
2、權限django
3、限制訪問頻率api
4、總結緩存
1、認證（補充）
認證請求頭app
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 from rest_framework.views import APIView 4 from rest_framework.response import Response 5 from rest_framework.authentication import BaseAuthentication 6 from rest_framework.permissions import BasePermission 7 8 from rest_framework.request import Request 9 from rest_framework import exceptions 10 11 token_list = [ 12 'sfsfss123kuf3j123', 13 'asijnfowerkkf9812', 14 ] 15 16 17 class TestAuthentication(BaseAuthentication): 18 def authenticate(self, request): 19 """ 20 用戶認證，若是驗證成功後返回元組： (用戶,用戶Token) 21 :param request: 22 :return: 23 None,表示跳過該驗證； 24 若是跳過了全部認證，默認用戶和Token和使用配置文件進行設置 25 self._authenticator = None 26 if api_settings.UNAUTHENTICATED_USER: 27 self.user = api_settings.UNAUTHENTICATED_USER() # 默認值爲：匿名用戶 28 else: 29 self.user = None 30 31 if api_settings.UNAUTHENTICATED_TOKEN: 32 self.auth = api_settings.UNAUTHENTICATED_TOKEN()# 默認值爲：None 33 else: 34 self.auth = None 35 (user,token)表示驗證經過並設置用戶名和Token； 36 AuthenticationFailed異常 37 """ 38 val = request.query_params.get('token') 39 if val not in token_list: 40 raise exceptions.AuthenticationFailed("用戶認證失敗") 41 42 return ('登陸用戶', '用戶token') 43 44 def authenticate_header(self, request): 45 """ 46 Return a string to be used as the value of the `WWW-Authenticate` 47 header in a `401 Unauthenticated` response, or `None` if the 48 authentication scheme should return `403 Permission Denied` responses. 49 """ 50 pass 51 52 53 class TestPermission(BasePermission): 54 message = "權限驗證失敗" 55 56 def has_permission(self, request, view): 57 """ 58 判斷是否有權限訪問當前請求 59 Return `True` if permission is granted, `False` otherwise. 60 :param request: 61 :param view: 62 :return: True有權限；False無權限 63 """ 64 if request.user == "管理員": 65 return True 66 67 # GenericAPIView中get_object時調用 68 def has_object_permission(self, request, view, obj): 69 """ 70 視圖繼承GenericAPIView，並在其中使用get_object時獲取對象時，觸發單獨對象權限驗證 71 Return `True` if permission is granted, `False` otherwise. 72 :param request: 73 :param view: 74 :param obj: 75 :return: True有權限；False無權限 76 """ 77 if request.user == "管理員": 78 return True 79 80 81 class TestView(APIView): 82 # 認證的動做是由request.user觸發 83 authentication_classes = [TestAuthentication, ] 84 85 # 權限 86 # 循環執行全部的權限 87 permission_classes = [TestPermission, ] 88 89 def get(self, request, *args, **kwargs): 90 # self.dispatch 91 print(request.user) 92 print(request.auth) 93 return Response('GET請求，響應內容') 94 95 def post(self, request, *args, **kwargs): 96 return Response('POST請求，響應內容') 97 98 def put(self, request, *args, **kwargs): 99 return Response('PUT請求，響應內容')
1 # 2 class MyAuthtication(BasicAuthentication): 3 def authenticate(self, request): 4 token = request.query_params.get('token') #注意是沒有GET的，用query_params表示 5 if token == 'zxxzzxzc': 6 return ('uuuuuu','afsdsgdf') #返回user，auth 7 # raise AuthenticationFailed('認證錯誤') #只要拋出認證錯誤這樣的異常就會去執行下面的函數 8 raise APIException('認證錯誤') 9 def authenticate_header(self, request): #認證不成功的時候執行 10 return 'Basic reala="api"' 11 12 class UserView(APIView): 13 authentication_classes = [MyAuthtication,] 14 def get(self,request,*args,**kwargs): 15 print(request.user) 16 print(request.auth) 17 return Response('用戶列表')
2、權限
一、需求：Host是匿名用戶和用戶都能訪問 #匿名用戶的request.user = none；User只有註冊用戶能訪問ide
1 from app03 import views 2 from django.conf.urls import url 3 urlpatterns = [ 4 # django rest framework 5 url('^auth/', views.AuthView.as_view()), 6 url(r'^hosts/', views.HostView.as_view()), 7 url(r'^users/', views.UsersView.as_view()), 8 url(r'^salary/', views.SalaryView.as_view()), 9 ]
1 class SalaryView(APIView): 2 '''用戶能訪問''' 3 message ='無權訪問' 4 authentication_classes = [MyAuthentication,] #驗證是否是用戶 5 permission_classes = [MyPermission,AdminPermission,] #再看用戶有沒有權限，若是有權限在判斷有沒有管理員的權限 6 def get(self,request): 7 return Response('薪資列表') 8 9 def permission_denied(self, request, message=None): 10 """ 11 If request is not permitted, determine what kind of exception to raise. 12 """ 13 if request.authenticators and not request.successful_authenticator: 14 '''若是沒有經過認證，而且權限中return False了，就會報下面的這個異常了''' 15 raise exceptions.NotAuthenticated(detail='無權訪問') 16 raise exceptions.PermissionDenied(detail=message)
1 from django.shortcuts import render 2 from rest_framework.views import APIView #繼承的view 3 from rest_framework.response import Response #友好的返回 4 from rest_framework.authentication import BaseAuthentication #認證的類 5 from rest_framework.authentication import BasicAuthentication 6 from rest_framework.permissions import BasePermission 7 from app01 import models 8 from rest_framework import exceptions 9 from rest_framework.permissions import AllowAny #權限在這個類裏面 10 from rest_framework.throttling import BaseThrottle,SimpleRateThrottle 11 # Create your views here. 12 # +++++++++++++++認證類和權限類======================== 13 class MyAuthentication(BaseAuthentication): 14 def authenticate(self, request): 15 token = request.query_params.get('token') 16 obj = models.UserInfo.objects.filter(token=token).first() 17 if obj : #若是認證成功，返回用戶名和auth 18 return (obj.username,obj) 19 return None #若是沒有認證成功就不處理，進行下一步 20 21 def authenticate_header(self, request): 22 pass 23 24 class MyPermission(BasePermission): 25 message = '無權訪問' 26 def has_permission(self,request,view): #has_permission裏面的self是view視圖對象 27 if request.user: 28 return True #若是不是匿名用戶就說明有權限 29 return False #不然無權限 30 31 class AdminPermission(BasePermission): 32 message = '無權訪問' 33 def has_permission(self, request, view): # has_permission裏面的self是view視圖對象 34 if request.user=='haiyun': 35 return True # 返回True表示有權限 36 return False #返回False表示無權限 37 38 # +++++++++++++++++++++++++++ 39 class AuthView(APIView): 40 authentication_classes = [] #認證頁面不須要認證 41 42 def get(self,request): 43 self.dispatch 44 return '認證列表' 45 46 class HostView(APIView): 47 '''需求： 48 Host是匿名用戶和用戶都能訪問 #匿名用戶的request.user = none 49 User只有註冊用戶能訪問 50 ''' 51 authentication_classes = [MyAuthentication,] 52 permission_classes = [] #都能訪問就不必設置權限了 53 def get(self,request): 54 print(request.user) 55 print(request.auth) 56 print(111111) 57 return Response('主機列表') 58 59 class UsersView(APIView): 60 '''用戶能訪問,request.user裏面有值''' 61 authentication_classes = [MyAuthentication,] 62 permission_classes = [MyPermission,AdminPermission] 63 def get(self,request): 64 print(request.user,'111111111') 65 return Response('用戶列表') 66 67 def permission_denied(self, request, message=None): 68 """ 69 If request is not permitted, determine what kind of exception to raise. 70 """ 71 if request.authenticators and not request.successful_authenticator: 72 '''若是沒有經過認證，而且權限中return False了，就會報下面的這個異常了''' 73 raise exceptions.NotAuthenticated(detail='無權訪問22222') 74 raise exceptions.PermissionDenied(detail=message) 75 76 77 class SalaryView(APIView): 78 '''用戶能訪問''' 79 message ='無權訪問' 80 authentication_classes = [MyAuthentication,] #驗證是否是用戶 81 permission_classes = [MyPermission,AdminPermission,] #再看用戶有沒有權限，若是有權限在判斷有沒有管理員的權限 82 def get(self,request): 83 return Response('薪資列表') 84 85 def permission_denied(self, request, message=None): 86 """ 87 If request is not permitted, determine what kind of exception to raise. 88 """ 89 if request.authenticators and not request.successful_authenticator: 90 '''若是沒有經過認證，而且權限中return False了，就會報下面的這個異常了''' 91 raise exceptions.NotAuthenticated(detail='無權訪問') 92 raise exceptions.PermissionDenied(detail=message)
若是趕上這樣的，還能夠自定製，參考源碼函數
def check_permissions(self, request): """ Check if the request should be permitted. Raises an appropriate exception if the request is not permitted. """ for permission in self.get_permissions(): #循環每個permission對象，調用has_permission #若是False，則拋出異常 #True 說明有權訪問 if not permission.has_permission(request, self): self.permission_denied( request, message=getattr(permission, 'message', None) )
def permission_denied(self, request, message=None): """ If request is not permitted, determine what kind of exception to raise. """ if request.authenticators and not request.successful_authenticator: '''若是沒有經過認證，而且權限中return False了，就會報下面的這個異常了''' raise exceptions.NotAuthenticated() raise exceptions.PermissionDenied(detail=message)
那麼咱們能夠重寫permission_denied這個方法，以下：源碼分析
1 class UsersView(APIView): 2 '''用戶能訪問,request.user裏面有值''' 3 authentication_classes = [MyAuthentication,] 4 permission_classes = [MyPermission,] 5 def get(self,request): 6 return Response('用戶列表') 7 8 def permission_denied(self, request, message=None): 9 """ 10 If request is not permitted, determine what kind of exception to raise. 11 """ 12 if request.authenticators and not request.successful_authenticator: 13 '''若是沒有經過認證，而且權限中return False了，就會報下面的這個異常了''' 14 raise exceptions.NotAuthenticated(detail='無權訪問') 15 raise exceptions.PermissionDenied(detail=message)
2. 全局使用post
上述操做中均是對單獨視圖進行特殊配置，若是想要對全局進行配置，則須要再配置文件中寫入便可。
1 REST_FRAMEWORK = { 2 'UNAUTHENTICATED_USER': None, 3 'UNAUTHENTICATED_TOKEN': None, #將匿名用戶設置爲None 4 "DEFAULT_AUTHENTICATION_CLASSES": [ 5 "app01.utils.MyAuthentication", 6 ], 7 'DEFAULT_PERMISSION_CLASSES':[ 8 "app03.utils.MyPermission",#設置路徑， 9 ] 10 }
1 class AuthView(APIView): 2 authentication_classes = [] #認證頁面不須要認證 3 4 def get(self,request): 5 self.dispatch 6 return '認證列表' 7 8 class HostView(APIView): 9 '''需求： 10 Host是匿名用戶和用戶都能訪問 #匿名用戶的request.user = none 11 User只有註冊用戶能訪問 12 ''' 13 authentication_classes = [MyAuthentication,] 14 permission_classes = [] #都能訪問就不必設置權限了 15 def get(self,request): 16 print(request.user) 17 print(request.auth) 18 return Response('主機列表') 19 20 class UsersView(APIView): 21 '''用戶能訪問,request.user裏面有值''' 22 authentication_classes = [MyAuthentication,] 23 permission_classes = [MyPermission,] 24 def get(self,request): 25 print(request.user,'111111111') 26 return Response('用戶列表') 27 28 def permission_denied(self, request, message=None): 29 """ 30 If request is not permitted, determine what kind of exception to raise. 31 """ 32 if request.authenticators and not request.successful_authenticator: 33 '''若是沒有經過認證，而且權限中return False了，就會報下面的這個異常了''' 34 raise exceptions.NotAuthenticated(detail='無權訪問') 35 raise exceptions.PermissionDenied(detail=message) 36 37 38 class SalaryView(APIView): 39 '''用戶能訪問''' 40 message ='無權訪問' 41 authentication_classes = [MyAuthentication,] #驗證是否是用戶 42 permission_classes = [MyPermission,AdminPermission,] #再看用戶有沒有權限，若是有權限在判斷有沒有管理員的權限 43 def get(self,request): 44 return Response('薪資列表') 45 46 def permission_denied(self, request, message=None): 47 """ 48 If request is not permitted, determine what kind of exception to raise. 49 """ 50 if request.authenticators and not request.successful_authenticator: 51 '''若是沒有經過認證，而且權限中return False了，就會報下面的這個異常了''' 52 raise exceptions.NotAuthenticated(detail='無權訪問') 53 raise exceptions.PermissionDenied(detail=message)
3、限流
一、爲何要限流呢?
答：
- - 第一點：爬蟲，反爬
- - 第二點：控制 API 訪問次數
- - 登陸用戶的用戶名能夠作標識
- 匿名用戶能夠參考 ip，可是 ip能夠加代理。.
只要你想爬，遲早有一天能夠爬。
二、限制訪問頻率源碼分析
1 self.check_throttles(request)
1 def check_throttles(self, request): 2 """ 3 Check if request should be throttled. 4 Raises an appropriate exception if the request is throttled. 5 """ 6 for throttle in self.get_throttles(): 7 #循環每個throttle對象，執行allow_request方法 8 # allow_request: 9 #返回False，說明限制訪問頻率 10 #返回True，說明不限制，通行 11 if not throttle.allow_request(request, self): 12 self.throttled(request, throttle.wait()) 13 #throttle.wait()表示還要等多少秒就能訪問了
1 def get_throttles(self): 2 """ 3 Instantiates and returns the list of throttles that this view uses. 4 """ 5 #返回對象 6 return [throttle() for throttle in self.throttle_classes]
1 throttle_classes = api_settings.DEFAULT_THROTTLE_CLASSES
1 class BaseThrottle(object): 2 """ 3 Rate throttling of requests. 4 """ 5 6 def allow_request(self, request, view): 7 """ 8 Return `True` if the request should be allowed, `False` otherwise. 9 """ 10 raise NotImplementedError('.allow_request() must be overridden') 11 12 def get_ident(self, request): 13 """ 14 Identify the machine making the request by parsing HTTP_X_FORWARDED_FOR 15 if present and number of proxies is > 0. If not use all of 16 HTTP_X_FORWARDED_FOR if it is available, if not use REMOTE_ADDR. 17 """ 18 xff = request.META.get('HTTP_X_FORWARDED_FOR') 19 remote_addr = request.META.get('REMOTE_ADDR') 20 num_proxies = api_settings.NUM_PROXIES 21 22 if num_proxies is not None: 23 if num_proxies == 0 or xff is None: 24 return remote_addr 25 addrs = xff.split(',') 26 client_addr = addrs[-min(num_proxies, len(addrs))] 27 return client_addr.strip() 28 29 return ''.join(xff.split()) if xff else remote_addr 30 31 def wait(self): 32 """ 33 Optionally, return a recommended number of seconds to wait before 34 the next request. 35 """ 36 return None
1 zz
1 def throttled(self, request, wait): 2 """ 3 If request is throttled, determine what kind of exception to raise. 4 """ 5 raise exceptions.Throttled(wait)
1 class Throttled(APIException): 2 status_code = status.HTTP_429_TOO_MANY_REQUESTS 3 default_detail = _('Request was throttled.') 4 extra_detail_singular = 'Expected available in {wait} second.' 5 extra_detail_plural = 'Expected available in {wait} seconds.' 6 default_code = 'throttled' 7 8 def __init__(self, wait=None, detail=None, code=None): 9 if detail is None: 10 detail = force_text(self.default_detail) 11 if wait is not None: 12 wait = math.ceil(wait) 13 detail = ' '.join(( 14 detail, 15 force_text(ungettext(self.extra_detail_singular.format(wait=wait), 16 self.extra_detail_plural.format(wait=wait), 17 wait)))) 18 self.wait = wait 19 super(Throttled, self).__init__(detail, code)
下面來看看最簡單的從源碼中分析的示例,這只是舉例說明了一下
1 from django.conf.urls import url 2 from app04 import views 3 urlpatterns = [ 4 url('limit/',views.LimitView.as_view()), 5 6 ]
1 from django.shortcuts import render 2 from rest_framework.views import APIView 3 from rest_framework.response import Response 4 from rest_framework import exceptions 5 # from rest_framewor import 6 # Create your views here. 7 class MyThrottle(object): 8 def allow_request(self,request,view): 9 #返回False，限制 10 #返回True，不限制 11 pass 12 def wait(self): 13 return 1000 14 15 16 class LimitView(APIView): 17 authentication_classes = [] #不讓認證用戶 18 permission_classes = [] #不讓驗證權限 19 throttle_classes = [MyThrottle, ] 20 def get(self,request): 21 # self.dispatch 22 return Response('控制訪問頻率示例') 23 24 def throttled(self, request, wait): 25 '''可定製方法設置中文錯誤''' 26 # raise exceptions.Throttled(wait) 27 class MyThrottle(exceptions.Throttled): 28 default_detail = '請求被限制' 29 extra_detail_singular = 'Expected available in {wait} second.' 30 extra_detail_plural = 'Expected available in {wait} seconds.' 31 default_code = '還須要再等{wait}秒' 32 raise MyThrottle(wait)
三、需求：對匿名用戶進行限制，每一個用戶一分鐘容許訪問10次（只針對用戶來講）
a、基於用戶IP限制訪問頻率
流程分析：
- 先獲取用戶信息，若是是匿名用戶，獲取IP。若是不是匿名用戶就能夠獲取用戶名。
- 獲取匿名用戶IP，在request裏面獲取，好比IP= 1.1.1.1。
- 吧獲取到的IP添加到到recode字典裏面，須要在添加以前先限制一下。
- 若是時間間隔大於60秒，說明時間久遠了，就把那個時間給剔除 了pop。在timelist列表裏面如今留的是有效的訪問時間段。
- 而後判斷他的訪問次數超過了10次沒有，若是超過了時間就return False。
- 美中不足的是時間是固定的，咱們改變他爲動態的：列表裏面最開始進來的時間和當前的時間進行比較，看須要等多久。
具體實現：
1 from django.shortcuts import render 2 from rest_framework.views import APIView 3 from rest_framework.response import Response 4 from rest_framework import exceptions 5 from rest_framework.throttling import BaseThrottle,SimpleRateThrottle #限制訪問頻率 6 import time 7 # Create your views here. 8 RECORD = {} 9 class MyThrottle(BaseThrottle): 10 11 def allow_request(self,request,view): 12 '''對匿名用戶進行限制，每一個用戶一分鐘訪問10次 ''' 13 ctime = time.time() 14 ip = '1.1.1.1' 15 if ip not in RECORD: 16 RECORD[ip] = [ctime] 17 else: 18 #[152042123,15204212,3152042,123152042123] 19 time_list = RECORD[ip] #獲取ip裏面的值 20 while True: 21 val = time_list[-1]#取出最後一個時間，也就是訪問最先的時間 22 if (ctime-60)>val: #吧時間大於60秒的給剔除了 23 time_list.pop() 24 #剔除了以後timelist裏面就是有效的時間了，在進行判斷他的訪問次數是否是超過10次 25 else: 26 break 27 if len(time_list) >10: 28 return False # 返回False，限制 29 time_list.insert(0, ctime) 30 return True #返回True，不限制 31 32 def wait(self): 33 ctime = time.time() 34 first_in_time = RECORD['1.1.1.1'][-1] 35 wt = 60-(ctime-first_in_time) 36 return wt 37 38 39 class LimitView(APIView): 40 authentication_classes = [] #不讓認證用戶 41 permission_classes = [] #不讓驗證權限 42 throttle_classes = [MyThrottle, ] 43 def get(self,request): 44 # self.dispatch 45 return Response('控制訪問頻率示例') 46 47 def throttled(self, request, wait): 48 '''可定製方法設置中文錯誤''' 49 # raise exceptions.Throttled(wait) 50 class MyThrottle(exceptions.Throttled): 51 default_detail = '請求被限制' 52 extra_detail_singular = 'Expected available in {wait} second.' 53 extra_detail_plural = 'Expected available in {wait} seconds.' 54 default_code = '還須要再等{wait}秒' 55 raise MyThrottle(wait)
1 # from django.shortcuts import render 2 # from rest_framework.views import APIView 3 # from rest_framework.response import Response 4 # from rest_framework import exceptions 5 # from rest_framework.throttling import BaseThrottle,SimpleRateThrottle #限制訪問頻率 6 # import time 7 # # Create your views here. 8 # RECORD = {} 9 # class MyThrottle(BaseThrottle): 10 # 11 # def allow_request(self,request,view): 12 # '''對匿名用戶進行限制，每一個用戶一分鐘訪問10次 ''' 13 # ctime = time.time() 14 # ip = '1.1.1.1' 15 # if ip not in RECORD: 16 # RECORD[ip] = [ctime] 17 # else: 18 # #[152042123,15204212,3152042,123152042123] 19 # time_list = RECORD[ip] #獲取ip裏面的值 20 # while True: 21 # val = time_list[-1]#取出最後一個時間，也就是訪問最先的時間 22 # if (ctime-60)>val: #吧時間大於60秒的給剔除了 23 # time_list.pop() 24 # #剔除了以後timelist裏面就是有效的時間了，在進行判斷他的訪問次數是否是超過10次 25 # else: 26 # break 27 # if len(time_list) >10: 28 # return False # 返回False，限制 29 # time_list.insert(0, ctime) 30 # return True #返回True，不限制 31 # 32 # def wait(self): 33 # ctime = time.time() 34 # first_in_time = RECORD['1.1.1.1'][-1] 35 # wt = 60-(ctime-first_in_time) 36 # return wt 37 # 38 # 39 # class LimitView(APIView): 40 # authentication_classes = [] #不讓認證用戶 41 # permission_classes = [] #不讓驗證權限 42 # throttle_classes = [MyThrottle, ] 43 # def get(self,request): 44 # # self.dispatch 45 # return Response('控制訪問頻率示例') 46 # 47 # def throttled(self, request, wait): 48 # '''可定製方法設置中文錯誤''' 49 # # raise exceptions.Throttled(wait) 50 # class MyThrottle(exceptions.Throttled): 51 # default_detail = '請求被限制' 52 # extra_detail_singular = 'Expected available in {wait} second.' 53 # extra_detail_plural = 'Expected available in {wait} seconds.' 54 # default_code = '還須要再等{wait}秒' 55 # raise MyThrottle(wait) 56 57 58 59 from django.shortcuts import render 60 from rest_framework.views import APIView 61 from rest_framework.response import Response 62 from rest_framework import exceptions 63 from rest_framework.throttling import BaseThrottle,SimpleRateThrottle #限制訪問頻率 64 import time 65 # Create your views here. 66 RECORD = {} 67 class MyThrottle(BaseThrottle): 68 69 def allow_request(self,request,view): 70 '''對匿名用戶進行限制，每一個用戶一分鐘訪問10次 ''' 71 ctime = time.time() 72 self.ip =self.get_ident(request) 73 if self.ip not in RECORD: 74 RECORD[self.ip] = [ctime] 75 else: 76 #[152042123,15204212,3152042,123152042123] 77 time_list = RECORD[self.ip] #獲取ip裏面的值 78 while True: 79 val = time_list[-1]#取出最後一個時間，也就是訪問最先的時間 80 if (ctime-60)>val: #吧時間大於60秒的給剔除了 81 time_list.pop() 82 #剔除了以後timelist裏面就是有效的時間了，在進行判斷他的訪問次數是否是超過10次 83 else: 84 break 85 if len(time_list) >10: 86 return False # 返回False，限制 87 time_list.insert(0, ctime) 88 return True #返回True，不限制 89 90 def wait(self): 91 ctime = time.time() 92 first_in_time = RECORD[self.ip][-1] 93 wt = 60-(ctime-first_in_time) 94 return wt 95 96 97 class LimitView(APIView): 98 authentication_classes = [] #不讓認證用戶 99 permission_classes = [] #不讓驗證權限 100 throttle_classes = [MyThrottle, ] 101 def get(self,request): 102 # self.dispatch 103 return Response('控制訪問頻率示例') 104 105 def throttled(self, request, wait): 106 '''可定製方法設置中文錯誤''' 107 # raise exceptions.Throttled(wait) 108 class MyThrottle(exceptions.Throttled): 109 default_detail = '請求被限制' 110 extra_detail_singular = 'Expected available in {wait} second.' 111 extra_detail_plural = 'Expected available in {wait} seconds.' 112 default_code = '還須要再等{wait}秒' 113 raise MyThrottle(wait)
b、用resetframework內部的限制訪問頻率（利於Django緩存）
源碼分析：
from rest_framework.throttling import BaseThrottle,SimpleRateThrottle #限制訪問頻率
1 class BaseThrottle(object): 2 """ 3 Rate throttling of requests. 4 """ 5 6 def allow_request(self, request, view): 7 """ 8 Return `True` if the request should be allowed, `False` otherwise. 9 """ 10 raise NotImplementedError('.allow_request() must be overridden') 11 12 def get_ident(self, request): #惟一標識 13 """ 14 Identify the machine making the request by parsing HTTP_X_FORWARDED_FOR 15 if present and number of proxies is > 0. If not use all of 16 HTTP_X_FORWARDED_FOR if it is available, if not use REMOTE_ADDR. 17 """ 18 xff = request.META.get('HTTP_X_FORWARDED_FOR') 19 remote_addr = request.META.get('REMOTE_ADDR') #獲取IP等 20 num_proxies = api_settings.NUM_PROXIES 21 22 if num_proxies is not None: 23 if num_proxies == 0 or xff is None: 24 return remote_addr 25 addrs = xff.split(',') 26 client_addr = addrs[-min(num_proxies, len(addrs))] 27 return client_addr.strip() 28 29 return ''.join(xff.split()) if xff else remote_addr 30 31 def wait(self): 32 """ 33 Optionally, return a recommended number of seconds to wait before 34 the next request. 35 """ 36 return None
1 class SimpleRateThrottle(BaseThrottle): 2 """ 3 一個簡單的緩存實現，只須要` get_cache_key() `。被覆蓋。 4 速率（請求/秒）是由視圖上的「速率」屬性設置的。類。該屬性是一個字符串的形式number_of_requests /期。 5 週期應該是：（的），「秒」，「M」，「min」，「h」，「小時」，「D」，「一天」。 6 之前用於節流的請求信息存儲在高速緩存中。 7 A simple cache implementation, that only requires `.get_cache_key()` 8 to be overridden. 9 10 The rate (requests / seconds) is set by a `rate` attribute on the View 11 class. The attribute is a string of the form 'number_of_requests/period'. 12 13 Period should be one of: ('s', 'sec', 'm', 'min', 'h', 'hour', 'd', 'day') 14 15 Previous request information used for throttling is stored in the cache. 16 """ 17 cache = default_cache 18 timer = time.time 19 cache_format = 'throttle_%(scope)s_%(ident)s' 20 scope = None 21 THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES 22 23 def __init__(self): 24 if not getattr(self, 'rate', None): 25 self.rate = self.get_rate() 26 self.num_requests, self.duration = self.parse_rate(self.rate) 27 28 def get_cache_key(self, request, view):#這個至關因而一個半成品，咱們能夠來補充它 29 """ 30 Should return a unique cache-key which can be used for throttling. 31 Must be overridden. 32 33 May return `None` if the request should not be throttled. 34 """ 35 raise NotImplementedError('.get_cache_key() must be overridden') 36 37 def get_rate(self): 38 """ 39 Determine the string representation of the allowed request rate. 40 """ 41 if not getattr(self, 'scope', None): 42 msg = ("You must set either `.scope` or `.rate` for '%s' throttle" % 43 self.__class__.__name__) 44 raise ImproperlyConfigured(msg) 45 46 try: 47 return self.THROTTLE_RATES[self.scope] 48 except KeyError: 49 msg = "No default throttle rate set for '%s' scope" % self.scope 50 raise ImproperlyConfigured(msg) 51 52 def parse_rate(self, rate): 53 """ 54 Given the request rate string, return a two tuple of: 55 <allowed number of requests>, <period of time in seconds> 56 """ 57 if rate is None: 58 return (None, None) 59 num, period = rate.split('/') 60 num_requests = int(num) 61 duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]] 62 return (num_requests, duration) 63 64 #一、一進來會先執行他， 65 def allow_request(self, request, view): 66 """ 67 Implement the check to see if the request should be throttled. 68 69 On success calls `throttle_success`. 70 On failure calls `throttle_failure`. 71 """ 72 if self.rate is None: 73 return True 74 75 self.key = self.get_cache_key(request, view) #二、執行get_cache_key，這裏的self.key就至關於咱們舉例ip 76 if self.key is None: 77 return True 78 79 self.history = self.cache.get(self.key, []) #三、獲得的key，默認是一個列表，賦值給了self.history， 80 # 這時候self.history就是每個ip對應的訪問記錄 81 self.now = self.timer() 82 83 # Drop any requests from the history which have now passed the 84 # throttle duration 85 while self.history and self.history[-1] <= self.now - self.duration: 86 self.history.pop() 87 if len(self.history) >= self.num_requests: 88 return self.throttle_failure() 89 return self.throttle_success() 90 91 def throttle_success(self): 92 """ 93 Inserts the current request's timestamp along with the key 94 into the cache. 95 """ 96 self.history.insert(0, self.now) 97 self.cache.set(self.key, self.history, self.duration) 98 return True 99 100 def throttle_failure(self): 101 """ 102 Called when a request to the API has failed due to throttling. 103 """ 104 return False 105 106 def wait(self): 107 """ 108 Returns the recommended next request time in seconds. 109 """ 110 if self.history: 111 remaining_duration = self.duration - (self.now - self.history[-1]) 112 else: 113 remaining_duration = self.duration 114 115 available_requests = self.num_requests - len(self.history) + 1 116 if available_requests <= 0: 117 return None 118 119 return remaining_duration / float(available_requests)
請求一進來會先執行SimpleRateThrottle這個類的構造方法
1 def __init__(self): 2 if not getattr(self, 'rate', None): 3 self.rate = self.get_rate() #點進去看到須要些一個scope ，2/m 4 self.num_requests, self.duration = self.parse_rate(self.rate)
1 def get_rate(self): 2 """ 3 Determine the string representation of the allowed request rate. 4 """ 5 if not getattr(self, 'scope', None): #檢測必須有scope，沒有就報錯了 6 msg = ("You must set either `.scope` or `.rate` for '%s' throttle" % 7 self.__class__.__name__) 8 raise ImproperlyConfigured(msg) 9 10 try: 11 return self.THROTTLE_RATES[self.scope] 12 except KeyError: 13 msg = "No default throttle rate set for '%s' scope" % self.scope 14 raise ImproperlyConfigured(msg)
1 def parse_rate(self, rate): 2 """ 3 Given the request rate string, return a two tuple of: 4 <allowed number of requests>, <period of time in seconds> 5 """ 6 if rate is None: 7 return (None, None) 8 num, period = rate.split('/') 9 num_requests = int(num) 10 duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]] 11 return (num_requests, duration)
1 #二、接下來會先執行他， 2 def allow_request(self, request, view): 3 """ 4 Implement the check to see if the request should be throttled. 5 6 On success calls `throttle_success`. 7 On failure calls `throttle_failure`. 8 """ 9 if self.rate is None: 10 return True 11 12 self.key = self.get_cache_key(request, view) #二、執行get_cache_key，這裏的self.key就至關於咱們舉例ip 13 if self.key is None: 14 return True #不限制 15 # [114521212,11452121211,45212121145,21212114,521212] 16 self.history = self.cache.get(self.key, []) #三、獲得的key，默認是一個列表，賦值給了self.history， 17 # 這時候self.history就是每個ip對應的訪問記錄 18 self.now = self.timer() 19 20 # Drop any requests from the history which have now passed the 21 # throttle duration 22 while self.history and self.history[-1] <= self.now - self.duration: 23 self.history.pop() 24 if len(self.history) >= self.num_requests: 25 return self.throttle_failure() 26 return self.throttle_success()
1 def wait(self): 2 """ 3 Returns the recommended next request time in seconds. 4 """ 5 if self.history: 6 remaining_duration = self.duration - (self.now - self.history[-1]) 7 else: 8 remaining_duration = self.duration 9 10 available_requests = self.num_requests - len(self.history) + 1 11 if available_requests <= 0: 12 return None 13 14 return remaining_duration / float(available_requests)
代碼實現：
1 ###########用resetframework內部的限制訪問頻率############## 2 class MySimpleRateThrottle(SimpleRateThrottle): 3 scope = 'xxx' 4 def get_cache_key(self, request, view): 5 return self.get_ident(request) #返回惟一標識IP 6 7 class LimitView(APIView): 8 authentication_classes = [] #不讓認證用戶 9 permission_classes = [] #不讓驗證權限 10 throttle_classes = [MySimpleRateThrottle, ] 11 def get(self,request): 12 # self.dispatch 13 return Response('控制訪問頻率示例') 14 15 def throttled(self, request, wait): 16 '''可定製方法設置中文錯誤''' 17 # raise exceptions.Throttled(wait) 18 class MyThrottle(exceptions.Throttled): 19 default_detail = '請求被限制' 20 extra_detail_singular = 'Expected available in {wait} second.' 21 extra_detail_plural = 'Expected available in {wait} seconds.' 22 default_code = '還須要再等{wait}秒' 23 raise MyThrottle(wait)
記得在settings裏面配置
1 REST_FRAMEWORK = { 2 'UNAUTHENTICATED_USER': None, 3 'UNAUTHENTICATED_TOKEN': None, #將匿名用戶設置爲None 4 "DEFAULT_AUTHENTICATION_CLASSES": [ 5 "app01.utils.MyAuthentication", 6 ], 7 'DEFAULT_PERMISSION_CLASSES':[ 8 # "app03.utils.MyPermission",#設置路徑， 9 ], 10 'DEFAULT_THROTTLE_RATES':{ 11 'xxx':'2/minute' #2分鐘 12 } 13 } 14 15 #緩存：放在文件 16 CACHES = { 17 'default': { 18 'BACKEND': 'django.core.cache.backends.filebased.FileBasedCache', 19 'LOCATION': 'cache', #文件路徑 20 } 21 }
四、對匿名用戶進行限制，每一個用戶1分鐘容許訪問5次，對於登陸的普通用戶1分鐘訪問10次，VIP用戶一分鐘訪問20次
- 好比首頁能夠匿名訪問
- #先認證，只有認證了才知道是否是匿名的，
- #權限登陸成功以後才能訪問， ，index頁面就不須要權限了
- If request.user #判斷登陸了沒有
1 from django.contrib import admin 2 3 from django.conf.urls import url, include 4 from app05 import views 5 6 urlpatterns = [ 7 url('index/',views.IndexView.as_view()), 8 url('manage/',views.ManageView.as_view()), 9 ]
1 from django.shortcuts import render 2 from rest_framework.views import APIView 3 from rest_framework.response import Response 4 from rest_framework.authentication import BaseAuthentication #認證須要 5 from rest_framework.throttling import BaseThrottle,SimpleRateThrottle #限流處理 6 from rest_framework.permissions import BasePermission 7 from rest_framework import exceptions 8 from app01 import models 9 # Create your views here. 10 ###############3##認證##################### 11 class MyAuthentcate(BaseAuthentication): 12 '''檢查用戶是否存在,若是存在就返回user和auth，若是沒有就返回''' 13 def authenticate(self, request): 14 token = request.query_params.get('token') 15 obj = models.UserInfo.objects.filter(token=token).first() 16 if obj: 17 return (obj.username,obj.token) 18 return None #表示我不處理 19 20 ##################權限##################### 21 class MyPermission(BasePermission): 22 message='無權訪問' 23 def has_permission(self, request, view): 24 if request.user: 25 return True #true表示有權限 26 return False #false表示無權限 27 28 class AdminPermission(BasePermission): 29 message = '無權訪問' 30 31 def has_permission(self, request, view): 32 if request.user=='haiyan': 33 return True # true表示有權限 34 return False # false表示無權限 35 36 ############3#####限流##################3## 37 class AnonThrottle(SimpleRateThrottle): 38 scope = 'wdp_anon' #至關於設置了最大的訪問次數和時間 39 def get_cache_key(self, request, view): 40 if request.user: 41 return None #返回None表示我不限制，登陸用戶我無論 42 #匿名用戶 43 return self.get_ident(request) #返回一個惟一標識IP 44 45 class UserThrottle(SimpleRateThrottle): 46 scope = 'wdp_user' 47 def get_cache_key(self, request, view): 48 #登陸用戶 49 if request.user: 50 return request.user 51 return None #返回NOne表示匿名用戶我無論 52 53 54 ##################視圖##################### 55 #首頁支持匿名訪問, 56 #無須要登陸就能夠訪問 57 class IndexView(APIView): 58 authentication_classes = [MyAuthentcate,] #認證判斷他是否是匿名用戶 59 permission_classes = [] #通常主頁就不須要權限驗證了 60 throttle_classes = [AnonThrottle,UserThrottle,] #對匿名用戶和普通用戶的訪問限制 61 62 def get(self,request): 63 # self.dispatch 64 return Response('訪問首頁') 65 66 def throttled(self, request, wait): 67 '''可定製方法設置中文錯誤''' 68 69 # raise exceptions.Throttled(wait) 70 class MyThrottle(exceptions.Throttled): 71 default_detail = '請求被限制' 72 extra_detail_singular = 'Expected available in {wait} second.' 73 extra_detail_plural = 'Expected available in {wait} seconds.' 74 default_code = '還須要再等{wait}秒' 75 76 raise MyThrottle(wait) 77 78 #需登陸就能夠訪問 79 class ManageView(APIView): 80 authentication_classes = [MyAuthentcate, ] # 認證判斷他是否是匿名用戶 81 permission_classes = [MyPermission,] # 通常主頁就不須要權限驗證了 82 throttle_classes = [AnonThrottle, UserThrottle, ] # 對匿名用戶和普通用戶的訪問限制 83 84 def get(self, request): 85 # self.dispatch 86 return Response('管理人員訪問頁面') 87 88 def throttled(self, request, wait): 89 '''可定製方法設置中文錯誤''' 90 91 # raise exceptions.Throttled(wait) 92 class MyThrottle(exceptions.Throttled): 93 default_detail = '請求被限制' 94 extra_detail_singular = 'Expected available in {wait} second.' 95 extra_detail_plural = 'Expected available in {wait} seconds.' 96 default_code = '還須要再等{wait}秒' 97 98 raise MyThrottle(wait)
4、總結
一、認證：就是檢查用戶是否存在；若是存在返回(request.user,request.auth)；不存在request.user/request.auth=None
二、權限：進行職責的劃分
三、限制訪問頻率
認證 - 類：authenticate/authenticate_header ##驗證不成功的時候執行的 - 返回值： - return None, - return (user,auth), - raise 異常 - 配置： - 視圖： class IndexView(APIView): authentication_classes = [MyAuthentication,] - 全局： REST_FRAMEWORK = { 'UNAUTHENTICATED_USER': None, 'UNAUTHENTICATED_TOKEN': None, "DEFAULT_AUTHENTICATION_CLASSES": [ # "app02.utils.MyAuthentication", ], } 權限 - 類：has_permission/has_object_permission - 返回值： - True、#有權限 - False、#無權限 - exceptions.PermissionDenied(detail="錯誤信息") #異常本身隨意，想拋就拋，錯誤信息本身指定 - 配置： - 視圖： class IndexView(APIView): permission_classes = [MyPermission,] - 全局： REST_FRAMEWORK = { "DEFAULT_PERMISSION_CLASSES": [ # "app02.utils.MyAuthentication", ], } 限流 - 類：allow_request/wait PS: scope = "wdp_user" - 返回值：
return True、#不限制
return False #限制 - 配置： - 視圖： class IndexView(APIView): throttle_classes=[AnonThrottle,UserThrottle,] def get(self,request,*args,**kwargs): self.dispatch return Response('訪問首頁') - 全局 REST_FRAMEWORK = { "DEFAULT_THROTTLE_CLASSES":[ ], 'DEFAULT_THROTTLE_RATES':{ 'wdp_anon':'5/minute', 'wdp_user':'10/minute', } }