标签：获取 lambda 判断 throttle first imp **kwargs 版本 当前时间
作用：验证用户是否登录
在视图类中写上authentication_classes = [ ]，这是一个列表
需要实现 authenticate() 方法
应用
自定义验证
import jwt
from rest_framework import exceptions
from rest_framework.authentication import BaseAuthentication
from rest_framework_jwt.settings import api_settings
from api import models
class HulaQueryParamAuthentication(BaseAuthentication):
def authenticate(self, request):
"""
# raise Exception(), 不在继续往下执行，直接返回给用户。
# return None ,本次认证完成，执行下一个认证
# return ('x',"x")，认证成功，不需要再继续执行其他认证了，继续往后权限、节流、视图函数
"""
token = request.query_params.get('token')
if not token:
raise exceptions.AuthenticationFailed({'code':10002,'error':"登录成功之后才能操作"})
jwt_decode_handler = api_settings.JWT_DECODE_HANDLER
try:
payload = jwt_decode_handler(token)
except jwt.ExpiredSignature:
raise exceptions.AuthenticationFailed({'code':10003,'error':"token已过期"})
except jwt.DecodeError:
raise exceptions.AuthenticationFailed({'code':10004,'error':"token格式错误"})
except jwt.InvalidTokenError:
raise exceptions.AuthenticationFailed({'code':10005,'error':"认证失败"})
jwt_get_username_from_payload = api_settings.JWT_PAYLOAD_GET_USERNAME_HANDLER
username = jwt_get_username_from_payload(payload)
user_object = models.UserInfo.objects.filter(username=username).first()
return (user_object,token)局部的配置
#局部配置（给需要的视图类写上即可）
# 登录用户才能访问所有注册用户列表(局部使用设置)
authentication_classes = [MyBaseAuthentication,]
#设置为空列表，就不走认证流程了（全局设置后，要想单个视图不走认证，让它等于一个空的列表）
authentication_classes = []全局的配置
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES':['util.authentication.MyBaseAuthentication',],
}
#匿名用户（用户未登录的情况下，可以设置匿名用户信息（ user、auth））
# 必要条件：不管是全局还是视图中都没有设置认证类
REST_FRAMEWORK = {
'UNAUTHENTICATED_USER':lambda :'匿名用户',
'UNAUTHENTICATED_TOKEN':lambda :'1234',
}实现流程
1. 当用户的请求进来时，执行APIView中的dispatch方法；
2. 这个方法中有一个initialize_request方法；
3. 方法的返回值是一个Request类实例化的对象，
Request(
request, # request在这个类中重新封装
parsers=self.get_parsers(), # 解析器
# 获取认证的类实例化的对象列表，
authenticators=self.get_authenticators(),
)
4. 然后在执行dispatch中的initial这个方法;
5. 这个方法中执行了self.perform_authentication(request)这个方法；
6. 这个方法中是request.user，这个方法时Request这个类中的一个属性方法；
7. 这属性方法中执行了_authenticate方法；
8. 这个方法中获取你在第3步中赋值给authenticators属性的对象列表，并且循环这个对象列表，中每个对象中的authenticate(self)这个方法；
作用：某些接口只能是特定的用户才能访问
在视图类中配置permission_classes = [] ,这是一个列表
需要实现 has_permission() 方法，表示全部对象对资源的权限问题
还有一个 has_object_permission()表示某个对象是否对资源有权限
返回 True 或 False，True 表示有权限，False 表示 无
全局配置 DEFAULT_PERMISSION_CLASSES，认证类中属性 message，表示无权限时返回的内容
权限自定义类最好继承 BasePermission
引用
自定义权限类
#自定义权限类（Object）,自定义的需要将它需要的
class MyPermission(object):
message = 'vip用户才能访问'
def has_permission(self,request,view):
#完成权限逻辑
#返回True,表示有权限访问
#返回Flase，表示没有权限访问
#自定义权限类，继承BasePermission类重构它的方法;
from rest_framework.permissions import BasePermission
class MyPermission(BasePermission):
message = 'vip用户才能访问'
def has_permission(self,request,view):
#完成权限逻辑
#返回True,表示有权限访问
#返回Flase，表示没有权限访问局部配置：
#设置权限类(局部使用设置)
permission_classes = [MyPermission,]
全局设置：
#全局设置
REST_FRAMEWORK = {
'DEFAULT_PERMISSION_CLASSES':['util.permission.MyPermission',],
{
#设置为空列表，就不走权限流程了（全局设置后，要想单个视图不走权限设置了）
permission_classes = []实现流程
1. 首先当用户将请求提交过来之后走APIView类中的dispatch方法,执行这个方法中的initial这个方法；
2. 在执行这个方法中的self.check_permissions(request)方法;
3. 这个方法中首先去获取权限类实例化的列表，然后循环这个列表，执行每个对象中的has_permission这个方法；作用：通过获取未登录用户的ip，或者以登录用户的属性来对他们在一定时间内的访问频率
在试图类中throttle_classes = [],写上这个列表，列表中写上节流的模块
如果是自己定义的类必须实现allow_request()这个方法
返回值成功True,失败False
应用
自定义：
#源码详解
RECORD_VISITS = {}
class VisitThrottle(object):
def __init__(self):
#获取用户的访问历史
self.history = []
def allow_request(self, request, view):
#allow_request是否允许方法
#True 允许访问
#False 不允许访问
#获取用户的IP地址
# ip_adress = request._request.META.get('REMOTE_ADDR')
# key = ip_adress
#基于用户
token = request.auth
key = token
currenttime = time.time()
if key not in RECORD_VISITS:
#当前的IP地址没有访问过服务器
RECORD_VISITS[key] = [currenttime]
return True
#获取访问历史记录
visit_history = RECORD_VISITS[key]
self.history = visit_history
#[ 12:01:00, 12:01:25, 12:02:25] 12:03:30 - 60
# 12:02:30
while visit_history and visit_history[-1] < currenttime - 60:
visit_history.pop()
if len(visit_history) < 5:
#每分钟访问5次
visit_history.insert(0,currenttime)
return True
return False # False表示访问频率太高被限制
def wait(self):
# 12:03:03
# [12:02:58,12:02:55,12:02:50,12:02:45,12:02:40]
first_time = self.history[-1]
return 60 - (time.time() - first_time)
#自定义（SimpleRateThrottle）
class MySimpleRateThrottle(SimpleRateThrottle):
scope = 'unlogin'
def get_cache_key(self, request, view):
#根据ip或者用户标示获取用户的访问记录
return self.get_ident(request)
# return request.user.name局部使用：
#局部使用
#设置节流的类(局部使用设置)
throttle_classes = [VisitThrottle,]
throttle_classes = [MySimpleRateThrottle,]
#设置为空列表，就不进行节流设置了
throttle_classes = []全局使用：
# 全局设置
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_RATES':{
'unlogin':'5/m',
},
'DEFAULT_THROTTLE_CLASSES':['util.throttle.MySimpleRateThrottle',],
}实现流程
1. 首先当用户将请求提交过来之后走APIView类中的dispatch方法,执行这个方法中的initial这个方法；
2. 在执行这个方法中的self.check_throttles(request)方法;
3. 这个方法中首先去获取权限类实例化的列表，然后循环这个列表，执行每个对象中的allow_request这个方法；
4. 这个方法中能获取当前用户的历史记录，如果当前用户没有访问记录就让它继续访问，如果有就判断当前历史记录中的最后一次记录的时间跟当前时间减去所规定时间的大小；如果当前时间减去规定时间后大于最后一次访问时间就让它继续访问；
5. 如果在成功之后会将这次的访问记录加到历史记录中；作用：获取版本信息，可以给用户返回不同版本的数据；
在视图类中加上versioning_class = QueryParameterVersioning,就可以当前的版本进行校验
需要在settings文件中配置"ALLOWED_VERSIONS":[‘v1‘,‘v2‘],,这个配置表示你当前允许的版本信息，只有在这个列表中的版本才有权利访问；
使用
# 源码解析：
class BaseVersioning:
# 设置默认版本
default_version = api_settings.DEFAULT_VERSION
# 设置允许的版本
allowed_versions = api_settings.ALLOWED_VERSIONS
# 设置版本参数
version_param = api_settings.VERSION_PARAM
def reverse(self, viewname, args=None, kwargs=None, request=None, format=None, **extra):
return _reverse(viewname, args, kwargs, request, format, **extra)
def is_allowed_version(self, version):
# version是当前的用户的版本信息
# 如果允许的版本中没有值
if not self.allowed_versions:
return True
# 否则会执行下边的判断，判断后返回一个值
return ((version is not None and version == self.default_version) or
(version in self.allowed_versions))
class URLPathVersioning(BaseVersioning):
def determine_version(self, request, *args, **kwargs):
# 获取当前用户的版本信息
version = kwargs.get(self.version_param, self.default_version)
# 判断是不是为空
if version is None:
# 如果为空就取默认版本信息
version = self.default_version
# 执行is_allowed_version方法，然后判断返回值，
if not self.is_allowed_version(version):
# 如果返回的是一个False就主动抛出异常
raise exceptions.NotFound(self.invalid_version_message)
return version
局部使用
from rest_framework.versioning import URLPathVersioning
# 在视图中（局部使用）
versioning_class = URLPathVersioning
# 如果那个是视图不需要写一个
versioning_class = None全局使用
#如果需要做版本的默认和限制，需要在settings中设置
REST_FRAMEWORK = {
'DEFAULT_VERSION':'v1',
'ALLOWED_VERSIONS':['v1','v2','v3'],
'VERSION_PARAM':'version',
}
#全局设置
REST_FRAMEWORK = {
'DEFAULT_VERSIONING_CLASS':'rest_framework.versioning.URLPathVersioning',
}实现流程
1. 首先当用户将请求提交过来之后走APIView类中的dispatch方法,执行这个方法中的initial这个方法；
2. 在执行这个方法中的self.determine_version(request, *args, **kwargs)方法;
3. 获取版本验证的类，实例化这个类，在执行这个类中的determine_version这个方法；
4. 这个方法会获取你当前url中的版本信息，然后调用is_allowed_version方法去判断当前这个版本信息是不是被允许，允许就继续执行；标签：获取 lambda 判断 throttle first imp **kwargs 版本 当前时间
原文地址：https://www.cnblogs.com/zhufanyu/p/11872898.html
