Day85--drf06--整体流程及源码分析、全局异常处理与接口文档
作者：互联网
昨日回顾
1 频率限制
-写一个类，继承SimpleRateThrottle，重写get_cache_key，
返回什么就以什么做限制（限制ip，限制用户id，手机号），
-再写一个类属性scope='字符串'， 需要跟配置文件中对应 '字符串'：'5/m'
-局部配置，全局配置
# 注意：若是以ip限制全局配置，那么所有的接口总计不能超过指定的次数，就不单是某个接口次数限制
2 过滤和排序
-内置过滤类：SearchFilter
-内置排序类：OrderingFilter
-配置在继承GenericAPIView+ListModelMixin及其子类的view接口中，类属性：filter_backends=[]
-配置相应的字段
-search_fields=['name']
-ordering_fields = ['price']
3 第三方过滤类
-配置在类属性上：filter_backends = [DjangoFilterBackend]
-配置字段：filterset_fields=['name','price']
-http://127.0.0.1:8000/books/?name=红楼梦&price=12
4 自定义过滤类
-自己写一个类，继承BaseFilterBackend，重写filter_queryset，返回queryset对象，queryset是过滤后的
-配置在继承GenericAPIView+ListModelMixin及其子类，类属性：filter_backends
5 分页功能
-三个分页类
PageNumberPagination：基本分页 ?page=1&size=2
-四个类属性
LimitOffsetPagination：偏移分页 ?limit=3&offset=2
-四个类属性
CursorPagination：游标分页 选择上一页和下一页
-三个类属性
-cursor_query_param = 'cursor' # 查询条件
-page_size = 2 # 每页显示多少条
-ordering = 'id' #按谁排序
-只需要配置在继承GenericAPIView+ListModelMixin及其子类，类属性：pagination_class
今日内容
1. drf整体流程
# drf处于的位置：路由匹配成功，进视图类之前
# 整体处理流程：
1.包装新的request对象
2.处理编码（urlencoded，formdata，json）
3.三大认证（顺序：认证、权限、频率）（*****）
4.进入视图类：
-4.1 若是继承（APIView）（*****）
-去模型中取数据
-序列化（*****）
-返回数据
-4.2 若是继承（GenericAPIView+ListModelMixin）（*****）
-去模型中取数据
-进行过滤和排序
-分页
-序列化（*****）
-返回数据
-5.处理了响应（浏览器，json）
-6.处理了全局异常
2. 源码分析--认证、频率、权限
2.1 认证源码
# 1.入口：APIView的dispatch()中--->self.initial()中处理三大认证--->认证类的代码：self.perform_authentication(request)
# 2.self.perform_authentication(request)--->就是APIView的perform_authentication()
def perform_authentication(self, request):
request.user # 执行了新的request对象的user方法
# 3.Request类的user方法
@property
def user(self):
if not hasattr(self, '_user'):
with wrap_attributeerrors():
# 核心就是这句话，是Request的_authenticate()方法
self._authenticate()
return self._user
# 4.Request类的_authenticate(self)方法 (核心)
def _authenticate(self):
for authenticator in self.authenticators: # self.authenticators是个列表，列表中放了一个个认证类的对象
try:
# self是request，所以自定义认证类的authenticate，有两个参数(self,request)，第二个参数request给了这里的self
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 # 解压赋值,后续的request对象就有user属性了
return
# 源码可知：若是写了多个认证类，只要有一个认证类中认证通过，返回了user和token，后续的认证类就不会for循环执行了
self._not_authenticated()
# 5.Request类的self.authenticators属性 是多个认证类对象的列表
-是在Request初始化的时候，传入的
-Request类是在什么时候初始化的---》APIView的dispatch中的刚开始位置
-APIView的dispatch()---> request = self.initialize_request(request, *args, **kwargs)
# 6.APIView的self.initialize_request方法
def initialize_request(self, request, *args, **kwargs):
return Request(
request,
parsers=self.get_parsers(),
authenticators=self.get_authenticators(),# APIView的
negotiator=self.get_content_negotiator(),
parser_context=parser_context
)
# 7.APIView的get_authenticators方法
def get_authenticators(self):
# 列表中放了一个个认证类的对象
return [auth() for auth in self.authentication_classes]
2.2 权限源码
# 1.入口：APIView的dispatch()中--->self.initial()中处理三大认证--->权限类的代码self.check_permissions(request)
# 2.APIView的check_permissions方法
def check_permissions(self, request):
for permission in self.get_permissions(): # 列表，是一个个视图类中配置的权限类的对象
if not permission.has_permission(request, self):
self.permission_denied(
request,
message=getattr(permission, 'message', None),
code=getattr(permission, 'code', None)
)
# 3.APIView的self.get_permissions():
def get_permissions(self):
# 列表里放了一个个权限类的对象
return [permission() for permission in self.permission_classes]
# 4.权限认证失败，返回中文
-在权限类中配置message即可（给对象，类都可以）
2.3 频率源码
# 1.入口：APIView的dispatch()中--->self.initial()中处理三大认证--->频率类的代码self.check_throttles(request)
# 2.APIView的self.check_throttles(request)
def check_throttles(self, request):
throttle_durations = []
for throttle in self.get_throttles(): # 列表，是一个个视图类中配置的频率类的对象
# 如果被限制了，就把剩余时间 追加到 throttle_durations 限制持续时间 列表中
if not throttle.allow_request(request, self):
throttle_durations.append(throttle.wait())
# 3.APIView的self.get_throttles()
def get_throttles(self):
# 列表里放了一个个频率类的对象
return [throttle() for throttle in self.throttle_classes]
# 从频率的源码可知：
若是自定义的频率类 是继承SimpleRateThrottle的，可以直接将限制的值 写在频率限制类中
THROTTLE_RATES = {'ip_m_3': '3/m'}
# SimpleRateThrottle源码分析：
def get_rate(self): # 根据配置的限制值的key:scope，从THROTTLE_RATES中 取出限制值：rate
"""
Determine the string representation of the allowed request rate.
"""
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] # scope：'user' => '3/min'
except KeyError:
msg = "No default throttle rate set for '%s' scope" % self.scope
raise ImproperlyConfigured(msg)
def parse_rate(self, rate): # 根据配置拆分出 限制次数 和 持续时间
"""
Given the request rate string, return a two tuple of:
<allowed number of requests>, <period of time in seconds>
"""
if rate is None:
return (None, None)
# 3 m
num, period = rate.split('/') # rate：'3/min'
num_requests = int(num)
duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
return (num_requests, duration)
def allow_request(self, request, view):
if self.rate is None:
return True
# 当前登录用户的ip地址
self.key = self.get_cache_key(request, view) # key：'throttle_user_1'
if self.key is None:
return True
# 从缓存中取出 存放IP访问时间 的列表，若是初次访问，缓存为空，self.history为[]
self.history = self.cache.get(self.key, [])
# 获取一下当前时间，存放到 self.now
self.now = self.timer()
# Drop any requests from the history which have now passed the throttle duration
# 将时间列表中 超过持续时间的 全部删除
# 当前访问与第一次访问时间间隔如果大于60s，将第一次记录清除，不再算作一次计数
# self.history:[10:23,10:55]
# now:10:56
while self.history and self.now - self.history[-1] >= self.duration:
self.history.pop()
# history的长度与限制次数3进行比较
# history 长度第一次访问0，第二次访问1，第三次访问2，第四次访问3失败
if len(self.history) >= self.num_requests:
# 直接返回False，代表频率限制了
return self.throttle_failure()
# history的长度未达到限制次数3，代表可以访问
return self.throttle_success()
def throttle_success(self): # 将当前时间插入到history列表的开头，将history列表、key、过期时间作为数据存到缓存中
"""
Inserts the current request's timestamp along with the key into the cache.
"""
self.history.insert(0, self.now)
self.cache.set(self.key, self.history, self.duration)
return True
3. 自定义全局异常处理
# 1.自定义一个全局异常处理函数
def common_exception_handler(exc, context):
"""
# 一般会将错误信息添加到日志：
通过参数 exc可获取错误提示信息，通过参数context可获取错误详细信息
str(exec) # 错误提示信息
str(context['view']) # 错误发生的视图
context['request'].META.get('REMOTE_ADDR') # IP地址
context['request'].user.id # 用户的ID
"""
# 1. 执行drf默认的异常处理
response = exception_handler(exc, context)
# 2.判断是否能被默认的异常处理捕获
if response:
# 2.1 若能被默认的捕获到，会返回一个Response对象出来，我们可.data 拿出原本默认要返回给前端的数据，再结合自己的格式 指定返回
return Response(data={'code': 9998, 'msg': response.data})
else:
# 2.2 若不能被默认的捕获到，会返回None，就自定义处理 返回给前端
return Response(data={'code': 9999, 'msg': '服务器异常，请联系系统管理员'})
# 2.在配置文件中
REST_FRAMEWORK = {
'EXCEPTION_HANDLER': 'app01.exceptions.common_exception_handler',
}
4. 自动生成接口文档
# 前后的分离
-前端一批人
-根本不知道你写了什么接口，请求参数什么样，响应数据什么样
-使用什么编码都不知道
-后端一批人
-我们写了很多接口
# 需要写接口文档（不同公司有规范）
-1 公司有接口文档平台，后端在平台上录入接口
-2 使用第三方接口文档平台，后端写了在平台录入
-Yapi:开源
-3 使用md，word文档写，写完传到git上
-4 自动生成接口文档（swagger，coreapi）
-通过swagger自动生成后导出，再导入到Yapi中
# coreapi 第三方模块--自动生成接口文档
# 1 安装：pip install coreapi
# 2 在路由中配置
from rest_framework.documentation import include_docs_urls
urlpatterns = [
...
path('docs/', include_docs_urls(title='站点页面标题'))
]
# 3 视图类：自动接口文档能生成的是继承自APIView及其子类的视图。
-1) 单一方法的视图，可直接使用类视图的文档字符串，如
class BookListView(generics.ListAPIView):
"""
返回所有图书信息.
"""
-2) 包含多个方法的视图，在类视图的文档字符串中，分开方法定义，如
class BookListCreateView(generics.ListCreateAPIView):
"""
get:
返回所有图书信息.
post:
新建图书.
"""
-3) 对于视图集ViewSet，仍在类视图的文档字符串中，分开定义，但是应使用action名称区分，如
class BookInfoViewSet(mixins.ListModelMixin, mixins.RetrieveModelMixin, GenericViewSet):
"""
list:
返回图书列表数据
retrieve:
返回图书详情数据
latest:
返回最新的图书数据
read:
修改图书的阅读量
"""
# 4 在配置文件中配置
REST_FRAMEWORK = {
'DEFAULT_SCHEMA_CLASS': 'rest_framework.schemas.coreapi.AutoSchema',
}
补充
1.函数显示传参类型和返回值
# python3.5以上版本，typing模块提高代码健壮性 (公司常见写的方式)
from typing import List, Tuple, Dict
def test(a: int, string: str, f: float, b: bool) -> Tuple[List, Tuple, Dict, bool]:
ll=[1,2,3,4]
tup = (string, a, string)
dic = {"xxx": f}
boo = b
return ll, tup, dic, boo
print(test(12, "lqz", 2.3, False))
作业
# 1 三个认证的源码---》自己捋一遍
# 2 写一个全局异常处理函数，保证无论出什么异常，前端都返回固定格式
def common_exception(exc, context):
# 第一步：记录异常信息到日志
print(
f'异常视图：{str(context["view"])} 访问IP：{context["request"].META.get("REMOTE_ADDR")} '
f'访问用户ID：{context["request"].user.id} 异常信息：{str(exc)}')
# 第二步：调用rest_framework 的异常捕获
response = exception_handler(exc, context)
# 第三步：判断是否能被内置的捕获到
if response:
# 若能，将内部返回的数据取出来，再加上我们自己的格式 返回给前端
return Response(data={'code': 9999, 'msg': response.data})
return Response(data={'code': 99998, 'msg': '系统出错，请联系管理员'})
# 3 试一下coreapi自动生成接口文档
# 4 给你一个地址，上地址看看Yapi怎么用
标签：return,get,--,self,request,源码,Day85,throttle,def 来源： https://www.cnblogs.com/Edmondhui/p/15680757.html