DRF節流組件

1.DRF節流組件自定義（限制訪問頻率） 　　

　方式一 自定義類和方法：

　　和上述的認證組件使用方式同樣，定義一個頻率組件類，推薦繼承BaseThrottle類，

　　需定義defallow_request(self,request,view):pass方法和defwait(self):pass提示信息方法

　　

 

 

    

 

 

 　seetings.py　

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'app01.apps.App01Config',
    'rest_framework',
]
 
MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    # 'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
 
REST_FRAMEWORK = {
    #認證配置（全局配置）----針對全部的繼承APIView的類，最終都會有結果返回
    'DEFAULT_AUTHENTICATION_CLASSES' : ['app01.utils.auth.MyAuth',],#能夠自定義多個認證類
    # 'UNAUTHENTICATED_USER':lambda :'匿名用戶request.user自定義值',#request.user有默認值
    # 'UNAUTHENTICATED_TOKEN':lambda :'request.auth自定義值',#request.auth有默認值
 
    #權限配置（全局配置）----針對全部的繼承APIView的類，在認證以後執行，沒有權限會返回message,有權限繼續執行
    'DEFAULT_PERMISSION_CLASSES':['app01.utils.permission.SVIPPermission',],#能夠自定義多個權限
 
 
    #頻率配置（全局配置）----針對全部的繼承APIView的類，在認證和權限校驗以後
    'DEFAULT_THROTTLE_CLASSES':['app01.utils.throttle.VisitThrottle',],#能夠定義多個訪問頻率類
}

settings.py

　　utils--auth.py--MyAuth認證類

 

#認證組件
from rest_framework.authentication import BaseAuthentication,BasicAuthentication
 
class MyAuth(BaseAuthentication):#能夠直接繼承BaseAuthentication類，能夠省略authenticate_header方法，或者繼承BasicAuthentication
    def authenticate(self, request):
                token = request._request.GET.get('token')
        token_obj = models.UserToken.objects.get(token=token)
        if not token_obj:
            raise exceptions.AuthenticationFailed('未認證用戶！！！')
        return (token_obj.user', 'request.auth')  # 認證函數執行結果若是經過則爲元組，元組第一個元素封裝在爲request.user
 

utils--auth.py--MyAuth認證類

　　utils--permission.py--SVIPPermission/MyPermission權限類 

from rest_framework.permissions import BasePermission
 
# 權限組件
class SVIPPermission(BasePermission):#推薦繼承BasePermission類
    # message = 'You do not have permission to perform this action.'#默認值
    message = '無此權限！！！'
    def has_permission(self,request, view):
        if request.user.user_type == 3:
            return False    #False爲沒權限
        return True   #True爲有權限
 
class MyPermission(BasePermission):#推薦繼承BasePermission類
    def has_permission(self,request, view):
        if request.user.user_type != 3:
            return False
        return True

utils--permission.py--SVIPPermission/MyPermission權限類

　　utils--throttle.py--VisitThrottle頻率類

 

'''
訪問頻率通常存儲在緩存或者數據庫中，以往程序重啓數據消失
再次示例使用字典存儲，經過ip進行節流演示
'''
import time
from rest_framework.throttling import BaseThrottle
 
# 節流組件
visit_record = {}
 
# 設置訪問頻率5次/60s
TIME_LIMIT = 60
NUM_LIMIT = 5
 
 
# class VisitThrottle(object):
class VisitThrottle(BaseThrottle):  # 推薦繼承BaseThrottle類
    '''
    自定義頻率組件類，推薦繼承BaseThrottle類，
    需定義 def allow_request(self, request, view):pass方法和def wait(self):pass提示信息方法
    '''
 
    def __init__(self):
        self.visit_history = []
 
    def allow_request(self, request, view):
 
        # remote_addr = request.META.get('REMOTE_ADDR')  # 獲取客戶端ip:request._request.META.get('REMOTE_ADDR')一樣功能
        remote_addr = self.get_ident(request)  # 父類BaseThrottle中已經實現了此方法能夠直接調用
 
        visit_time = time.time()
        if remote_addr not in visit_record:  # 當前ip訪問記錄爲空是第一次訪問
            visit_record[remote_addr] = [visit_time]
            return True
        else:
            visit_history = visit_record.get(remote_addr)
            while visit_history and (visit_time - visit_history[-1]) > TIME_LIMIT:  # 訪問記錄存在而且最先時間記錄與本次間隔大於限制時間就刪除
                visit_history.pop()
            if len(visit_history) < NUM_LIMIT:  # 當前訪問記錄次數與設定值比較，若是小於限定次數便可訪問
                visit_history.insert(0, visit_time)
                return True
        # 訪問記錄與本次訪問時間封裝在對象中以便wait調用
        self.visit_history = visit_record.get(remote_addr)
        self.visit_time = visit_time
 
        return False  # False表示超次數
 
    def wait(self):
        '''
        超出次數提示信息
        :return: 提示信息:秒數        '''
 
        return TIME_LIMIT - (self.visit_time - self.visit_history[-1])

utils--throttle.py--VisitThrottle頻率類

　　models.py

from django.db import models
# Create your models here
class UserInfo(models.Model):
    """
    用戶表
    """
    user_type_choices = [
        (1, '普通用戶'),
        (2, 'VIP用戶'),
        (3, 'SVIP用戶'),
    ]
    user_type = models.IntegerField(choices=user_type_choices)
    username = models.CharField(max_length=10, unique=True)
    password = models.CharField(max_length=12, null=False)
 
 
class UserToken(models.Model):
    """
    token表
    """
 
    user = models.OneToOneField(to='UserInfo')
    token = models.CharField(max_length=64)
    create_time = models.DateTimeField(auto_now=True)
 
 
class Book(models.Model):
    name = models.CharField(max_length=12)
 

models.py

　　urls.py

 

from django.conf.urls import url
from django.contrib import admin
from app01 import views
 
urlpatterns = [
    url(r'^admin/', admin.site.urls),
    url(r'^api/v1/login/$', views.AuthView.as_view()),
    url(r'^book/$', views.BookView.as_view(),name='book'),
    url(r'^order/$', views.OrderView.as_view(),name='order'),
 
]

urls.py

　　views.py

 

from django.shortcuts import render, HttpResponse
from rest_framework.views import APIView
from app01 import models
from django.http import JsonResponse
from app01.utils.auth import MyAuth
from app01.utils.permission import MyPermission,SVIPPermission
from app01.utils.throttle import VisitThrottle
from django.views import View
 
# Create your views here.
 
# 實例url:http://127.0.0.1:8000/book/?token=1
class BookView(APIView):
    # # （1）認證組件(局部使用)
    # authentication_classes = [MyAuth, ]
    #（2）權限組件（局部使用）
    permission_classes = [MyPermission,]
    # （3）頻率組件（局部使用）
    # throttle_classes = [VisitThrottle, ]
 
    def get(self, request):
        print(request.user)  # request.user在APIViewD的dispatch中進行封裝的
        return HttpResponse('GET')
 
    def post(self, request):
        return HttpResponse('POST')
 
    def put(self, request):
        return HttpResponse('PUT')
 
    def patch(self, request):
        return HttpResponse('PATCH')
 
    def delete(self, request):
        return HttpResponse('DELETE')
 
 
class OrderView(APIView):
    # （1）認證組件（局部使用）
    # authentication_classes = [MyAuth, ]
    # （2）權限組件（局部使用）
    # permission_classes = [SVIPPermission]
    #（3）頻率組件（局部使用）
    # throttle_classes = [VisitThrottle,]
 
    def get(self, request):
        print(request.user)  # request.user在認證組件中進行封裝的
        return HttpResponse('GET')
 
    def post(self, request):
        return HttpResponse('POST')
 
    def put(self, request):
        return HttpResponse('PUT')
 
    def patch(self, request):
        return HttpResponse('PATCH')
 
    def delete(self, request):
        return HttpResponse('DELETE')
 
 
import time
import hashlib
 
 
def token_md5(username):
    """
    自定義token
    :param username:
    :return:
    """
    t = time.time()
    md5 = hashlib.md5(str(t).encode('utf-8'))
    md5.update(username.encode('utf-8'))
    return md5.hexdigest()
 
 
class AuthView(View):
    #若是不註冊自定義組件，走默認的認證，最後返回了request.user和request.auth都是匿名用戶默認值，能夠在settings.py中加載自定義配置
    #可是在權限認證時不很差處理，因此仍是直接繼承View
    def post(self, request):
        """
        用戶登陸
        :param request:進行封裝以後的request對象
        :return: 登陸結果信息
        """
        ret = {'code': 0, 'msg': ''}
 
        username = request.POST.get('username', None)
        password = request.POST.get('password', None)
 
        # 每次登錄若是有就更新沒有就建立
        try:
            user_obj = models.UserInfo.objects.filter(username=username, password=password).first()
            if user_obj:
                token = token_md5(username)
                print(token)
                # 每次登錄若是有就更新沒有就建立
                models.UserToken.objects.update_or_create(user=user_obj, defaults={'token': token})
                ret['msg'] = '登錄成功！'
                ret['token'] = token
            else:
                ret['code'] = 1
                ret['msg'] = '帳號或密碼有誤！！！'
 
        except Exception as e:
            ret['code'] = 2
            ret['msg'] = '未知錯誤！！！'
        finally:
            return JsonResponse(ret)

views.py

 

2. DRF節流組件簡單配置方式全局

　　繼承SimpleRateThrottle類能夠直接進行簡單配置便可，無需本身實現節流方法

　　實現：scope和defget_cache_key(self,request,view):pass

　　示例中的utils--throttle.py--VisitThrottle頻率類方式二實現方式（結合settings.py配置）

　　

 

 

 　　settings.py

 

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'app01.apps.App01Config',
    'rest_framework',
]
 
MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    # 'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
REST_FRAMEWORK = {
    #認證配置（全局配置）----針對全部的繼承APIView的類，最終都會有結果返回（認證列表是或條件）
    'DEFAULT_AUTHENTICATION_CLASSES' : ['app01.utils.auth.MyAuth',],#能夠自定義多個認證類
    # 'UNAUTHENTICATED_USER':lambda :'匿名用戶request.user自定義值',#request.user有默認值
    # 'UNAUTHENTICATED_TOKEN':lambda :'request.auth自定義值',#request.auth有默認值
 
    #權限配置（全局配置）----針對全部的繼承APIView的類，在認證以後執行，沒有權限會返回message,有權限繼續執行
    'DEFAULT_PERMISSION_CLASSES':['app01.utils.permission.SVIPPermission',],#能夠自定義多個權限
 
 
    #頻率配置（全局配置）----針對全部的繼承APIView的類，在認證和權限校驗以後（節流校驗是並列的條件）
    'DEFAULT_THROTTLE_CLASSES':['app01.utils.throttle.VisitThrottle','app01.utils.throttle.UserThrottle',],#能夠定義多個訪問頻率類
    'DEFAULT_THROTTLE_RATES':{          #繼承節流組件的SimpleRateThrottle類使用
        'ThrottleTest':'5/m',#該key是在自定義的組件類定義的，value值形如：'5/s'或者'5/seconds'都可（只要是以s,m,h,d便可）
        #'LoginedUser':'10/m',#能夠針對不懂得身份標識進行節流規則制定
    }
}

settings.py

　　utils--auth.py--MyAuth認證類

 

#認證組件
from rest_framework.authentication import BaseAuthentication,BasicAuthentication
 
class MyAuth(BaseAuthentication):#能夠直接繼承BaseAuthentication類，能夠省略authenticate_header方法，或者繼承BasicAuthentication
    def authenticate(self, request):
                token = request._request.GET.get('token')
        token_obj = models.UserToken.objects.get(token=token)
        if not token_obj:
            raise exceptions.AuthenticationFailed('未認證用戶！！！')
        return (token_obj.user', 'request.auth')  # 認證函數執行結果若是經過則爲元組，元組第一個元素封裝在爲request.user
 

utils--auth.py--MyAuth認證類

　　utils--permission.py--SVIPPermission/MyPermission權限類 

from rest_framework.permissions import BasePermission
 
# 權限組件
class SVIPPermission(BasePermission):#推薦繼承BasePermission類
    # message = 'You do not have permission to perform this action.'#默認值
    message = '無此權限！！！'
    def has_permission(self,request, view):
        if request.user.user_type == 3:
            return False    #False爲沒權限
        return True   #True爲有權限
 
class MyPermission(BasePermission):#推薦繼承BasePermission類
    def has_permission(self,request, view):
        if request.user.user_type != 3:
            return False
        return True

utils--permission.py--SVIPPermission/MyPermission權限類

　　utils--throttle.py--VisitThrottle頻率類　　

'''
訪問頻率通常存儲在緩存或者數據庫中，以往程序重啓數據消失
再次示例使用字典存儲，經過ip進行節流演示
'''
import time
from rest_framework.throttling import BaseThrottle, SimpleRateThrottle
 
# 方式一節流組件（繼承BaseThrottle類）----徹底自定義寫法
"""
visit_record = {}
 
# 設置訪問頻率5次/60s
TIME_LIMIT = 60
NUM_LIMIT = 5
 
 
# class VisitThrottle(object):
class VisitThrottle(BaseThrottle):  # 推薦繼承BaseThrottle類
    '''
    自定義頻率組件類，推薦繼承BaseThrottle類，
    需定義 def allow_request(self, request, view):pass方法和def wait(self):pass提示信息方法
    '''
 
    def __init__(self):
        self.visit_history = []
 
    def allow_request(self, request, view):
 
        # remote_addr = request.META.get('REMOTE_ADDR')  # 獲取客戶端ip:request._request.META.get('REMOTE_ADDR')一樣功能
        remote_addr = self.get_ident(request)  # 父類BaseThrottle中已經實現了此方法能夠直接調用
 
        visit_time = time.time()
        if remote_addr not in visit_record:  # 當前ip訪問記錄爲空是第一次訪問
            visit_record[remote_addr] = [visit_time]
            return True
        else:
            visit_history = visit_record.get(remote_addr)
            while visit_history and (visit_time - visit_history[-1]) > TIME_LIMIT:  # 訪問記錄存在而且最先時間記錄與本次間隔大於限制時間就刪除
                visit_history.pop()
            if len(visit_history) < NUM_LIMIT:  # 當前訪問記錄次數與設定值比較，若是小於限定次數便可訪問
                visit_history.insert(0, visit_time)
                return True
        # 訪問記錄與本次訪問時間封裝在對象中以便wait調用
        self.visit_history = visit_record.get(remote_addr)
        self.visit_time = visit_time
 
        return False  # False表示超次數
 
    def wait(self):
        '''
        超出次數提示信息
        :return: 提示信息:秒數        '''
 
        return TIME_LIMIT - (self.visit_time - self.visit_history[-1])
 
"""
 
 
# 方式二節流組件（繼承SimpleRateThrottle類）----須要在全局直接配置訪問頻率
# 該方式直接使用rest_framework自帶的緩存機制,指定配置訪問頻率，還需重寫def get_cache_key(self, request, view):pass返回用戶訪問身份標識key
class VisitThrottle(SimpleRateThrottle):
    scope = 'ThrottleTest'  # 繼承SimpleRateThrottle類實現節流必須指定scope的值以便在全局配置使用
 
    def get_cache_key(self, request, view):
        return self.get_ident(request)#經過ip或代理進行節流
   
    
class UserThrottle(SimpleRateThrottle):
    '''
    本節流類以登陸用戶名爲標識
    '''
    scope = 'LoginedUser'  
 
    def get_cache_key(self, request, view):     
        return request.user#也能夠經過用戶名進行節流

utils--throttle.py--VisitThrottle頻率類

　　models.py　　

from django.db import models
# Create your models here
class UserInfo(models.Model):
    """
    用戶表
    """
    user_type_choices = [
        (1, '普通用戶'),
        (2, 'VIP用戶'),
        (3, 'SVIP用戶'),
    ]
    user_type = models.IntegerField(choices=user_type_choices)
    username = models.CharField(max_length=10, unique=True)
    password = models.CharField(max_length=12, null=False)
 
 
class UserToken(models.Model):
    """
    token表
    """
 
    user = models.OneToOneField(to='UserInfo')
    token = models.CharField(max_length=64)
    create_time = models.DateTimeField(auto_now=True)
 
 
class Book(models.Model):
    name = models.CharField(max_length=12)
 

models.py

　　urls.py

from django.conf.urls import url
from django.contrib import admin
from app01 import views
 
urlpatterns = [
    url(r'^admin/', admin.site.urls),
    url(r'^api/v1/login/$', views.AuthView.as_view()),
    url(r'^book/$', views.BookView.as_view(),name='book'),
    url(r'^order/$', views.OrderView.as_view(),name='order'),
 
]

urls.py

　　views.py　　

from django.shortcuts import render, HttpResponse
from rest_framework.views import APIView
from app01 import models
from django.http import JsonResponse
from app01.utils.auth import MyAuth
from app01.utils.permission import MyPermission,SVIPPermission
from app01.utils.throttle import VisitThrottle
from django.views import View
 
# Create your views here.
 
# 實例url:http://127.0.0.1:8000/book/?token=1
class BookView(APIView):
    # # （1）認證組件(局部使用)
    # authentication_classes = [MyAuth, ]
    #（2）權限組件（局部使用）
    permission_classes = [MyPermission,]
    # （3）頻率組件（局部使用）
    # throttle_classes = [VisitThrottle, ]
 
    def get(self, request):
        print(request.user)  # request.user在APIViewD的dispatch中進行封裝的
        return HttpResponse('GET')
 
    def post(self, request):
        return HttpResponse('POST')
 
    def put(self, request):
        return HttpResponse('PUT')
 
    def patch(self, request):
        return HttpResponse('PATCH')
 
    def delete(self, request):
        return HttpResponse('DELETE')
 
 
class OrderView(APIView):
    # （1）認證組件（局部使用）
    # authentication_classes = [MyAuth, ]
    # （2）權限組件（局部使用）
    # permission_classes = [SVIPPermission]
    #（3）頻率組件（局部使用）
    # throttle_classes = [VisitThrottle,]
 
    def get(self, request):
        print(request.user)  # request.user在認證組件中進行封裝的
        return HttpResponse('GET')
 
    def post(self, request):
        return HttpResponse('POST')
 
    def put(self, request):
        return HttpResponse('PUT')
 
    def patch(self, request):
        return HttpResponse('PATCH')
 
    def delete(self, request):
        return HttpResponse('DELETE')
 
 
import time
import hashlib
 
 
def token_md5(username):
    """
    自定義token
    :param username:
    :return:
    """
    t = time.time()
    md5 = hashlib.md5(str(t).encode('utf-8'))
    md5.update(username.encode('utf-8'))
    return md5.hexdigest()
 
 
class AuthView(View):
    #若是不註冊自定義組件，走默認的認證，最後返回了request.user和request.auth都是匿名用戶默認值，能夠在settings.py中加載自定義配置
    #可是在權限認證是很差經過，因此仍是直接繼承View
    def post(self, request):
        """
        用戶登陸
        :param request:進行封裝以後的request對象
        :return: 登陸結果信息
        """
        ret = {'code': 0, 'msg': ''}
 
        username = request.POST.get('username', None)
        password = request.POST.get('password', None)
 
        # 每次登錄若是有就更新沒有就建立
        try:
            user_obj = models.UserInfo.objects.filter(username=username, password=password).first()
            if user_obj:
                token = token_md5(username)
                print(token)
                # 每次登錄若是有就更新沒有就建立
                models.UserToken.objects.update_or_create(user=user_obj, defaults={'token': token})
                ret['msg'] = '登錄成功！'
                ret['token'] = token
            else:
                ret['code'] = 1
                ret['msg'] = '帳號或密碼有誤！！！'
 
        except Exception as e:
            ret['code'] = 2
            ret['msg'] = '未知錯誤！！！'
        finally:
            return JsonResponse(ret)
 

views.py