今日学习内容
一、权限类的使用
使用步骤：
-
写一个类，继承
BasePermission -
重写
has_permission方法 -
在方法中校验用户是否有权限（
request.user）就是当前登录用户 -
有权限，返回
True。没有权限，返回False。 -
self.message是给前端的提示信息。 -
局部配置，全局配置，局部禁用
局部配置： permission_classes = [UserTypePermission, ] 全局配置： "DEFAULT_PERMISSION_CLASSES":["app01.permission.UserTypePermission", ], 局部禁用： permission_classes = []
permission.py
from rest_framework.permissions import BasePermission
class UserTypePermission(BasePermission):
def has_permission(self, request, view):
# 只有超级管理员有权限
if request.user.user_type == 1:
return True # 有权限
else:
# self.message = '普通会员和免费会员没有权限 #返回前端的提示信息
'''
使用choice后， user.user_type 拿到的是数字类型，想要变成字符串
user.get_user_type_display()
'''
self.message = '你是: %s 用户, 你没有权限' % request.user.get_user_type_display()
return False # 没有权限
二、频率类的使用
无论是否登录是否有权限，都要限制访问的频率，比如一分钟访问3次。
使用步骤：
-
写一个类：继承
SimpleRateThrottle -
重写
get_cache_key, 返回唯一的字符串，会以这个字符串做频率限制(用户id或请求头中的ip地址) -
写一个类属性scop = '字符串', 必须要跟配置文件对象
-
配置文件中写
'DEFAULT_THROTTLE_RATES': { '随意写': '3/m' # 3/h 3/s 3/d } -
局部配置，全局配置，局部禁用。
局部配置： throttle_classes = [XThrottle, ] 全局配置： 'DEFAULT_THROTTLE_RATES': { 'xz': '5/m', 'xc': '3/m', } 局部禁用： throttle_classes = []
throttle.py
from rest_framework.throttling import SimpleRateThrottle, BaseThrottle
class XThrottle(SimpleRateThrottle):
# 类属性,这个类属性可以随意命名，但要跟配置文件对应
scope = 'xz'
def get_cache_key(self, request, view):
# 返回什么，频率就以什么做限制
# 可以通过用户id限制
# 可以通过ip地址限制(请求头里面可获取ip)
return request.META.get('REMOTE_ADDR')
.
'DEFAULT_THROTTLE_RATES': {
'xz': '5/m',
'xc': '3/m',
}
研究三大认证前要先研究API
API:
APIView的执行流程>>>入口：APIView的dispach
1.包装了新的request
2.执行了3大认证
3.执行视图类的方法
4.处理了全局异常
.
def dispatch(self, request, *args, **kwargs):
self.args = args
self.kwargs = kwargs
#第一步对request进行加工(添加数据) 请求模块
request = self.initialize_request(request, *args, **kwargs)
self.request = request
self.headers = self.default_response_headers
try:
#第二步： 处理版权信息 认证 权限 请求用户进行访问频率的限制 三大认证模块
self.initial(request, *args, **kwargs)
# Get the appropriate handler method
if request.method.lower() in self.http_method_names:
handler = getattr(self, request.method.lower(),
self.http_method_not_allowed)
else:
handler = self.http_method_not_allowed
#第三步：执行:get/post/put/delete函数 响应模块
response = handler(request, *args, **kwargs)
except Exception as exc:
#此处抛出异常 是针对第二步出现的错误 异常模块
response = self.handle_exception(exc)
#第四步: 对返回结果进行再次加工， 渲染模块
self.response = self.finalize_response(request, response, *args, **kwargs)
return self.response
.
APIView的dispatch中的497行：
self.initial(request, *args, **kwargs)
表示的是认证，权限，频率，
.
def initial(self, request, *args, **kwargs):
...
# 认证
\#\
self.perform_authentication(request)
# 权限
self.check_permissions(request)
# 请求用户访问频率限制
self.check_throttles(request)
三、认证源代码分析
def perform_authentication(self, request):
request.user
"""
request.user 为新的request，Resquest类中找user属性（方法），发现user是一个方法，包装成一个属性了。
"""
@property
def user(self):
if not hasattr(self, '_user'):
# Request类的对象中反射_user
with wrap_attributeerrors():
# 第一次会走这个代码
self._authenticate()
return self._user
def _authenticate(self):
for authenticator in self.authenticators:
# 配置在视图类中所有的认证类的对象列表
try:
# (user_token.user, token)
user_auth_tuple = authenticator.authenticate(self) # 调用认证类对象的authenticate
except exceptions.APIException:
self._not_authenticated()
raise
if user_auth_tuple is not None:
self._authenticator = authenticator # 忽略
self.user, self.auth = user_auth_tuple # 解压赋值
return
# 认证类可以配置多个，但是如果有个返回值有两个，后续就不会执行。
self._not_authenticated()
"""
总结：认证类，要重写authenticate方法，认证通过返回两个值或为None，认证不通过抛AuthenticationFailed（继承APIException）异常。
"""
四、权限源代码分析
def check_permissions(self, request):
for permission in self.get_permissions():
# permission 是配置在视图类中权限类的对象，对象调用它的绑定方法has_permission
# 对象调用自己绑定方法会把自己传入（权限类的对象，request， 视图类对象）
if not permission.has_permission(request, self):
self.permission_denied(
request,
message=getattr(permission, 'message', None),
code=getattr(permission, 'code', None)
)
def get_permissions(self):
"""
Instantiates and returns the list of permissions that this view requires.
"""
return [permission() for permission in self.permission_classes]
"""
self.permission_classes 视图类中配的权限类的列表
get_permissions 返回视图类中配的权限类的对象列表
eg:
[UserTypePermession(),]
"""
总结：
权限类源码：
为什么要写一个类，重写has_permission方法，有三个参数。
return True 或 False
if not permission.has_permission(request, self):
...
重写后的返回值要是 False才能走这个 if语句，如果是自动抛出异常是不会走这个 if语句的。
message可以做什么？
可以修改APIView里面捕获全局异常返回给前端的异常信息。
五、简单读频率类源码
def check_throttles(self, request):
throttle_durations = []
for throttle in self.get_throttles():
if not throttle.allow_request(request, self):
throttle_durations.append(throttle.wait())
总结：
如果要写频率类，继承BaseThrottle就要重写 allow_request方法，返回 True (没有到频率的限制) 或 False (到了频率的限制)
频率类源码具体分析
def check_throttles(self, request):
throttle_durations = [] #这个列表是用来存放下一次访问还需等待多长时间的
# 1）遍历配置的频率认证类，初始化得到一个个频率认证类对象（会调用频率认证类的 __init__() 方法）
# 2）频率认证类对象调用 allow_request 方法，判断是否限次（没有限次可访问，限次不可访问）
# 3）频率认证类对象在限次后，调用 wait 方法，获取还需等待多长时间可以进行下一次访问
# 注：频率认证类都是继承 SimpleRateThrottle 类
for throttle in self.get_throttles():
if not throttle.allow_request(request, self):
# 只要频率限制了，allow_request 返回False了，才会调用wait
throttle_durations.append(throttle.wait())
if throttle_durations:
# Filter out `None` values which may happen in case of config / rate
# changes, see #1438
durations = [
duration for duration in throttle_durations
if duration is not None
]
duration = max(durations, default=None)
self.throttled(request, duration)
.
'self.get_throttles() 就是视图类中频率类的对象列表'
def get_throttles(self):
return [throttle() for throttle in self.throttle_classes]
'''
将 self.throttle_classes中的 throttle遍历出来后 加括号实例化成对象存放在列表中 throttle()
throttle_classes = api_settings.DEFAULT_THROTTLE_CLASSES #说明它是在settings.py 中的 APISettings配置好的类'
'''
. SimpleRateThrottle
class SimpleRateThrottle(BaseThrottle):
cache = default_cache
timer = time.time
cache_format = 'throttle_%(scope)s_%(ident)s'
scope = None
THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES
def __init__(self):
#通过反射看自己类中是否有rate属性 没有就调用get_rate()方法 将返回值赋给rate
if not getattr(self, 'rate', None):
self.rate = self.get_rate()
self.num_requests, self.duration = self.parse_rate(self.rate)
# 紧接着通过parse_rate()方法 得到返回值 解压赋值给 num_requests(请求访问次数) duration(请求频率限制的时间)
.
def get_rate(self):
if not getattr(self, 'scope', None):
msg = ("You must set either `.scope` or `.rate` for '%s' throttle" %
self.__class__.__name__)
raise ImproperlyConfigured(msg)
try:
return self.THROTTLE_RATES[self.scope] #需要从settings.py的THROTTLE_RATES字典中获取值
这里的scope是继承SimpleRateThrottle的类中的写好的属性 例如UserRateThrottle中 scope = 'user'
注意：
throttling.py 文件中一共有五个类， 最基本的类是 BaseThrolle, 其次是 SimpleRateThrottle， 其他三个都继承 SimpleRateThrottle。
六、鸭子类型
# 指的是面向对中，子类不需要显示的继承某个类，只要有某个的方法和属性，那我就属于这个类
# 假设有个鸭子类Duck类，有两个方法，run，speak方法
# 假设又有一个普通鸭子类,PDuck，如果它也是鸭子，它需要继承Duck类，只要继承了鸭子类，什么都不需要写，普通鸭子类的对象就是鸭子这种类型；如果不继承，普通鸭子类的对象就不是鸭子这种类型
#假设又有一个唐老鸭子类,TDuck，如果它也是鸭子，它需要继承Duck类，只要继承了鸭子类，什么都不需要写，唐老鸭子类的对象就是鸭子这种类型；如果不继承，唐老鸭子类的对象就不是鸭子这种类型
总结：
以上解释对于其他的编程语言来说，继承后才能被称为被继承类的对象。
但是python不推崇这个，python推崇的是鸭子类型，指的是：
'只要这个类有被继承类里面的方法或者属性，那这个类就是被继承类的对象'
还要注意的是：
如果python鸭子类型的写法写错了，会有问题。
为了解决这个问题，就整出了一个强制性继承，必须要写里面的方法才能执行。
* abc模块，装饰后，必须要重写talk方法，不重写就会报错。
* drf源码中使用：父类中写这个方法，继承后没有具体实现也会报错，就抛出异常。
注意：
# django的配置文件不要乱导入，乱导入可能会出错
'''
django 的运行是在加载完配置后才能运行。
因为模块的导入会执行那个模块，而这个模块中又有别的导入，别的导入必须django运行起来才能使用。
'''
作业
# 1 编写图书和出版社的5个接口，所有接口都要有一分钟访问5次的频率限制
# 2 图书的接口需要登录才能方法，出版社的接口需要登录，并且是超级用户才能访问
# 3 整理认证和频率的执行流程
---------可以不需要写----
# 3 继承BaseThrottle编写频率类
# 自定义的逻辑 {ip:[时间1，时间2]}
#（1）取出访问者ip
#（2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走
#（3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间，
#（4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
#（5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败
.
from rest_framework.throttling import SimpleRateThrottle, BaseThrottle
class XThrottle(SimpleRateThrottle):
# 类属性,这个类属性可以随意命名，但要跟配置文件对应
scope = 'xz'
def get_cache_key(self, request, view):
# 返回什么，频率就以什么做限制
# 可以通过用户id限制
# 可以通过ip地址限制(请求头里面可获取ip)
return request.META.get('REMOTE_ADDR')
import time
class ZThrottle(BaseThrottle):
scope = 'xc'
user_ip = {}
def allow_request(self, request, view):
t = time.time()
# print(request.META)
ip = request.META.get('REMOTE_ADDR')
if ip not in self.user_ip:
self.user_ip[ip] = [t]
return True
self.ip_list = self.user_ip.get(ip)
while self.ip_list and t - self.ip_list[-1] > 60:
self.ip_list.pop()
if len(self.ip_list) < 3:
self.ip_list.insert(0, t)
return True
else:
return False
def wait(self):
t1 = time.time()
wait_time = 60 - (t1 - self.ip_list[-1])
return wait_time
