权限
rest_framework 执行dispatch-> self.check_permissions(request)方法
从settings文件获取指定的权限类列表,然后实例化对象。
settings中指定:DEFAULT_PERMISSION_CLASSES:[]
然后调用类的has_permission方法,对权限进行验证
自定义权限认证:
class Mypermission(BasePermission):
message = '错误'
def has_permission(self, request, view):
token = request.GET.get("token")
if token == '1':
return True
elif token == '2':
view.permission_denied(
request, message=getattr(Mypermission, 'message', None)
)
else:
return False
内置的权限认证类:
AllowAny , IsAuthenticated(django内置的权限判断is_authenticated方法判断) IsAdminUser IsAuthenticatedOrReadOnly等
限流
同样进入视图函数前,initial方法:
执行self.check_throttles(request)
settings文件全局添加: DEFAULT_THROTTLE_CLASSES
限流判断,执行对象的allow_request方法,当返回False时候,执行wait()
方法,判断多少秒后可访问。
自定义限流方法:
例如每个用户 10s 不能超过5次:
user_dict = {}
class BankAuth(BaseAuthentication):
def authenticate(self, request):
user = request._request.GET.get("user", None)
if not user:
raise AuthenticationFailed
return (user, None)
def authenticate_header(self, request):
pass
class Mythrottle(BaseThrottle):
def __init__(self):
self.history = None
def allow_request(self, request, view):
user = request.user
if user not in user_dict:
user_dict[user] = dict(name=user, view_count=1, last_time=time())
return True
else:
second = int(time() - user_dict[user]['last_time'])
if second < 10:
if user_dict[user]['view_count'] > 5:
self.history = user_dict[user]['last_time']
return False
else:
user_dict[user]['view_count'] = 0
user_dict[user]['last_time'] = time()
user_dict[user]['view_count'] += 1
return True
def wait(self):
return 10 - (time() - self.last_time)
class BankView(APIView):
authentication_classes = [BankAuth, ]
throttle_classes = [Mythrottle, ]
def get(self, request):
return JsonResponse(user_dict[request.user])
内置的SimpleRateThrottle
方法
# 首先执行allow_request方法
# 内部调用get_cache_key获取唯一标识别。待我们自己实现
# 'DEFAULT_THROTTLE_RATES': {
# 'my_throttle': '5/m' # ('s', 'sec', 'm', 'min', 'h', 'hour', 'd', #'day')
# }
class Mythrottle(SimpleRateThrottle):
scope = 'my_throttle' #settings文件中 DEFAULT_THROTTLE_RATES 配置的访问次数限制
def get_cache_key(self, request, view):
ident = self.get_ident(request) # 调用父类的获取唯一标识ip 方法
print(ident)
return ident