第85天-drf 06-整体流程和源代码分析、全局异常处理和接口文档

昨日回顾

1频率极限

-写一个类，继承SimpleRateThrottle，覆盖get_cache_key，

限制返回的内容(限制ip、限制用户id、移动电话号码)，

-编写另一个类属性范围=“字符串”，它需要对应于配置文件中的“字符串”:“5/m”

-本地配置，全局配置

#注意：如果全局配置受ip限制，那么所有接口的总数不能超过指定的次数，这意味着它不仅仅是一定数量的接口。

2过滤和排序

-内置过滤器类：搜索过滤器

-内置排序类：排序过滤器

-在继承genericpiviewlistmodelmixin及其子类的视图接口中配置，类属性：filter _ backs=[]。

-配置相应的字段。

-search_fields=['name']

-ordering_fields=['price']

3第三方过滤类

-在类属性上配置：filter _ back ends=[djangoofilterback]

-配置字段：filter set _ field=[' name '，' price']

-http://127 . 0 . 0 . 1:8000/书/名=红楼梦价格=12

4自定义过滤器类别

-自己编写一个类，继承basefilter后端，覆盖filter_queryset，返回queryset对象，该对象经过过滤。

-配置为继承genericpiviewlistmodelmixin及其子类，类属性：filter_backends

5分页功能

-三个分页类

页码分页：基本分页页面=1size=2

-四个类别属性

限制偏移量分页：偏移量分页限制=3偏移量=2

-四个类别属性

光标分页：光标分页选择上一页和下一页。

-三类属性

-cursor_query_param='cursor' #查询条件

-page_size=2 #显示多少页？

-订购='id' #由谁订购

-只需要配置为继承genericpiviewlistmodelmixin及其子类，class属性：paging _ class。

今日内容

1. drf整体流程

# drf的位置：在进入视图类之前，路线匹配成功。

#整体处理流程：

1.包装新的请求对象。

2.处理编码(urlencoded、formdata、json)

3.三大认证(顺序：认证、权威和频率)(* * * * * *)

4.输入视图类：

-4.1在继承的情况下(API view)(* * * * * *)

-从模型中获取数据。

-序列化(* * * * *)

-返回数据

-4.2在继承的情况下(genericpivot listmodemmixin)(* * * *)

-从模型中获取数据。

-过滤和分类

-分页

-序列化(* * * * *)

-返回数据

-5.处理响应(浏览器、json)

-6.处理全局异常

2. 源码分析--认证、频率、权限

2.1 认证源码

# 1.入口：APIView-self . initial()-身份验证类：self.perform _ authentication(请求)的dispatch()中三个身份验证类的代码

# 2.自我。execute _authentication(请求)-即执行APIView的_ authentication()。

def execute _ authentication(self，request):

Request.user #实现新请求对象的user方法。

# 3.请求类的用户方法

@property

def用户(自我):

如果没有，请拨打：

with wrap _ attributeerrors():

#核心是这句话，是Request的_authenticate()方法。

自我。验证(_ e)

()
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怎么用