zoukankan      html  css  js  c++  java
  • 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类可以直接进行简单配置即可,无需自己实现节流方法

      实现:scopedefget_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
  • 相关阅读:
    [模板] 循环数组的最大子段和
    [最短路][几何][牛客] [国庆集训派对1]-L-New Game
    [洛谷] P1866 编号
    1115 Counting Nodes in a BST (30 分)
    1106 Lowest Price in Supply Chain (25 分)
    1094 The Largest Generation (25 分)
    1090 Highest Price in Supply Chain (25 分)
    树的遍历
    1086 Tree Traversals Again (25 分)
    1079 Total Sales of Supply Chain (25 分 树
  • 原文地址:https://www.cnblogs.com/open-yang/p/11573239.html
Copyright © 2011-2022 走看看