目录
- 1.原生django的cookie+session认证底层原理
- 2.断点调试
- 3.认证，权限，频率源码分析
- 3.基于APIView来写分页功能
- 4.异常处理
1.原生django的cookie+session认证底层原理
2.断点调试
程序是debug模式运行，可以在任意位置停下，查看当前情况下变量数据的变化情况
3.认证，权限，频率源码分析
1.权限的源码执行流程
我们自定制了一个权限类，通过配置视图类中的permission_classes局部使用时，就会执行权限类的has_permission方法，完成权限校验
（1）APIView的dispatch中 - 3大认证
APIView的 dispatch 方法，在执行视图类的方法之前，会先执行3大认证【认证、权限、频率】的校验。
也就是在497行左右，在self.initial中进行
def dispatch(self, request, *args, **kwargs):
...
try:
self.initial(request, *args, **kwargs)
（2）APIView的 initial - 3大认证
在 【initial】 方法中，执行3大认证
def initial(self, request, *args, **kwargs):
"""
Runs anything that needs to occur prior to calling the method handler.
"""
# 能够解析的编码、版本控制
self.format_kwarg = self.get_format_suffix(**kwargs)
# Perform content negotiation and store the accepted info on the request
neg = self.perform_content_negotiation(request)
request.accepted_renderer, request.accepted_media_type = neg
# Determine the API version, if versioning is in use.
version, scheme = self.determine_version(request, *args, **kwargs)
request.version, request.versioning_scheme = version, scheme
"认证组件"
self.perform_authentication(request)
"权限组件"
self.check_permissions(request)
"频率组件"
self.check_throttles(request)
（3）执行APIView中的check_permissions方法 - 执行登录认证
-
self.get_permissions()是去执行我们在视图类上配置的【权限类列表】==>[CommonPermission(),] -
permission是我们写的【权限类的对象】 -
在没有权限的情况下，执行
permission.has_permission(request, self)，self是【视图类的对象】，就是自己写的权限类的has_permission的view参数 -
如果return的是False，就会走
self.permission_denied(...)，返回message和code结束 -
所以如果配置了多个权限类，前面的没有通过则不会执行下一个权限类了
def check_permissions(self, request):
"""
Check if the request should be permitted.
Raises an appropriate exception if the request is not permitted.
"""
# self.get_permissions()
# permission是我们写的权限类的对象，self.get_permissions()
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)
)
self是视图类的对象，是自己写的has_permission的view参数
（4）APIView的get_permissions方法
self.permission_classes 是我们自己在视图类上的列表，里面是一个个的权限类
-
self.permission_classes
-
permission_classes = [CommonPermission]
-
[CommonPermission()] 本质返回了权限类的对象，放在列表中
def get_permissions(self):
"""
Instantiates and returns the list of permissions that this view requires.
"""
# 1 self.permission_classes
# 2 permission_classes = [CommonPermission]
# 3 [CommonPermission()] 本质返回了权限类的对象，放在列表中
return [permission() for permission in self.permission_classes]
总结执行流程：
1.局部配置权限类时
权限类 -- APIView --- dispatch --- initial --- check_permissions(request)
从中取出配置在视图类上的权限类，实例化得到对象，一个个执行对象的has_permission方法，如果返回False就直接结束，不在往下继续执行了，权限认证结束
2.局部没有配置权限类，则会使用全局的配置
如果视图类上不配置权限类：permission_classes = [CommonPermission
-->会使用配置文件的api_settings.DEFAULT_PERMISSION_CLASSES
优先级： 局部>全局>默认
优先使用项目配置，其次使用默认配置【当全局中有的时候，会替换默认配置】
2.认证
（1）APIView中dispatch方法【同权限】
APIView的 dispatch 方法，在执行视图类的方法之前，会先执行3大认证【认证、权限、频率】的校验。
也就是在497行左右，在self.initial中进行
（2）APIView的 initial - 3大认证
在initial 方法中，执行3大认证的认证
# APIView类的399行左右：
def initial(self, request, *args, **kwargs):
# 能够解析的编码，版本控制。。。。
self.format_kwarg = self.get_format_suffix(**kwargs)
neg = self.perform_content_negotiation(request)
request.accepted_renderer, request.accepted_media_type = neg
version, scheme = self.determine_version(request, *args, **kwargs)
request.version, request.versioning_scheme = version, scheme
# 认证组件的执行位置【读它】
self.perform_authentication(request)
# 权限组件
self.check_permissions(request)
# 频率组件
self.check_throttles(request)
（3）APIView的 perform_authentication 方法 - 执行权限认证
request.user 是伪装成属性的方法，这里的request是drf包装了的新的request对象，所以我们去读drf的Request源码
def perform_authentication(self, request):
request.user
（4）Request类的user方法
user方法通过伪装可以以属性的方式使用
其中执行了 self._authenticate()方法
@property
def user(self):
"""
Returns the user associated with the current request, as authenticated
by the authentication classes provided to the request.
"""
if not hasattr(self, '_user'):
with wrap_attributeerrors():
self._authenticate()
return self._user
（5）Request类的_authenticate方法
self是Request的对象，self.authenticators是我们配置在视图类中放在列表中的一个个认证类对象，是在Request类初始化的时候，传入的- 返回了两个值，第一个是【当前用户】，第二个是【token】，当返回这两个值的时候后面的认证类就不继续进行了
-
其中解压赋值
self.user=当前登录用户，self.auth=token -
self是当次请求的【新的Request的对象】，所以登录认证完成后，通过request.user即可获得当前登录的用户对象
- 如果返回的是None，会继续执行下一个认证类
def _authenticate(self):
"""
Attempt to authenticate the request using each authentication instance
in turn.
"""
# 1 self.authenticators 我们配置在视图类上认证类的一个个对象，放到列表中
# Request类初始化的时候，传入的
for authenticator in self.authenticators:
try:
# 2 返回user_auth_tuple
user_auth_tuple = authenticator.authenticate(self)
# APIException是drf自己定义的一个异常
except exceptions.APIException:
self._not_authenticated()
raise
if user_auth_tuple is not None:
"解压赋值"
# self.user=当前登录用户，self是当次请求的新的Request的对象
# self.auth=token
self._authenticator = authenticator
self.user, self.auth = user_auth_tuple
return self._not_authenticated()
（6） Request 的__init__传入authenticators
- 通过上面的
self.authenticators，我们发现Request类在初始化的时候，自动传入了一个authenticators - 就是在APIView 的
dispatch中的492行，保存了产生了新的request对象
def __init__(self, request, parsers=None, authenticators=None,
negotiator=None, parser_context=None):
assert isinstance(request, HttpRequest), (
'The `request` argument must be an instance of '
'`django.http.HttpRequest`, not `{}.{}`.'
.format(request.__class__.__module__, request.__class__.__name__)
)
self.authenticators = authenticators or ()
if force_user is not None or force_token is not None:
forced_auth = ForcedAuthentication(force_user, force_token)
self.authenticators = (forced_auth,)
（7）APIView 的 dispatch
APIView 的 dispatch 中的492行，保存了产生了新的request对象，其中执行了initialize_request方法
def dispatch(self, request, *args, **kwargs):
request = self.initialize_request(request, *args, **kwargs)
self.request = request
（8）APIView 的 initialize_request
在initialize_request中，执行了authenticators=self.get_authenticators()，也就是初始化传入了authenticators
def initialize_request(self, request, *args, **kwargs):
"""
Returns the initial request object.
"""
parser_context = self.get_parser_context(request)
return Request(
request,
parsers=self.get_parsers(),
authenticators=self.get_authenticators(),
negotiator=self.get_content_negotiator(),
parser_context=parser_context
)
总结
1.配置在视图类上的认证类，会在执行视图类方法之前执行，并且在权限认证之前执行
2.自己定制的认证类，可以返回两个值则直接结束，返回None则继续执行下一个认证类
3.认证类中返回了当前登录用户，则可以在后续权限中可以通过request.user拿到当前登录用户
3.频率
（1）APIView中dispatch方法
【同上】
（2）APIView的 initial - 3大认证
【同上】
（3）APIView 的 check_throttles方法 - 执行频率认证
self.get_throttles()就是产生我们在视图类中局部配置的频率类的对象，并将对象放在列表中
每次取出一个频率类的对象，执行allow_request方法，如果是False，频率超了不再继续执行；是True则可以往后执行
def check_throttles(self, request):
"""
Check if request should be throttled.
Raises an appropriate exception if the request is throttled.
"""
throttle_durations = []
# self.get_throttles() 就是产生我们在视图类中局部配置的频率类的对象，并将对象放在列表中
# 每次取出一个频率类的对象，执行allow_request方法，如果是False，频率超了，不能再走了
for throttle in self.get_throttles():
# 如果是True，那么没有超出频率，继续执行
if not throttle.allow_request(request, self):
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)
总结
我们写的频率类，继承BaseThrottle，重写allow_request
在内部判断，如果超频了，就返回False，不允许访问；如果没超频率，就返回True运行继续访问
4.继承BaseThrottle自定义编写频率类
1.判断是否超频的逻辑
（1）取出访问者ip
（2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走 {ip地址:[时间1，时间2，时间3，时间4]}
（3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间，
（4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
（5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败
2.代码演示 - 通过模仿SimpleRateThrottle
VISIT_RECORD为存放ip地址对应访问时间的字典，key为ip，value为列表__init__初始化一个存放allow_request方法是，继承BaseThrottle需要重写的类，其中构建判断频率的逻辑，返回True为运行访问，返回False为限制访问wait方法返回一个时间，当有该方法存在时候，会显示限制访问的剩余时间
class SuperThrottle(BaseThrottle):
# 定义一个访问字典，存放ip地址对应的每次的访问时间
VISIT_RECORD = {}
# 定义一个记录登录时间的列表，初始化为None
def __init__(self):
self.history = None
def allow_request(self, request, view):
# （1）取出访问者ip
ip = request.META.get('REMOTE_ADDR')
# 导入时间
import time
ctime = time.time()
# （2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走 {ip地址:[时间1，时间2，时间3，时间4]}
if ip not in self.VISIT_RECORD:
self.VISIT_RECORD[ip] = [ctime]
return True
# 记录当前用户的访问时间，有则获取值
self.history = self.VISIT_RECORD.get(ip, [])
# （3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间，
while self.history and ctime - self.history[-1] > 60:
self.history.pop()
# （4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
if len(self.history) < 3:
self.history.insert(0, ctime)
return True
# （5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败
else:
return False
# wait 方法 清晰等待时间
def wait(self):
"""
Returns the recommended next request time in seconds.
返回一个单位为秒的建议等待时间
"""
import time
ctime = time.time()
return 60 - (ctime - self.history[-1])
5.SimpleRateThrottle源码分析
写一个频率类继承SimpleRateThrottle，其中的allow_request方法，来控制频率
def allow_request(self, request, view):
# self.rate是频率的限制，没有self.rate则没有限制，直接返回None
if self.rate is None:
return True
self.key = self.get_cache_key(request, view)
if self.key is None:
return True
self.history = self.cache.get(self.key, [])
self.now = self.timer()
# Drop any requests from the history which have now passed the
# throttle duration
while self.history and self.history[-1] <= self.now - self.duration:
self.history.pop()
if len(self.history) >= self.num_requests:
return self.throttle_failure()
return self.throttle_success()
self.rate是频率的限制，没有self.rate则没有限制，直接返回None，其通过配置文件中的scope参数来获取频率，没有scope的话则直接抛出异常
``THROTTLE_RATES是在配置文件中配置的频率限制，THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES` 5/h
-
SimpleRateThrottle的__init__方法得到
self.rate后，传给self.parse_rate方法，并解压复制给self.num_requests频率限制的次数,self.duration频率限制的时间
def __init__(self):
if not getattr(self, 'rate', None):
self.rate = self.get_rate()
self.num_requests, self.duration = self.parse_rate(self.rate)
-
SimpleRateThrottle的parse_rate方法通过切割解压赋值，得到
self.num_requests = num频率限制的次数,self.duration = duration频率限制的时间
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)
num, period = rate.split('/')
num_requests = int(num)
duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
return (num_requests, duration)
- 继续
SimpleRateThrottle的allow_request方法
def allow_request(self, request, view):
...
# 获取缓存中的 访问时间列表
self.history = self.cache.get(self.key, [])
# 获取现在的时间
self.now = self.timer()
# Drop any requests from the history which have now passed the
# throttle duration
# 比对访问时间差是否在限制的时间段内
while self.history and self.history[-1] <= self.now - self.duration:
# 如果 访问时间没有超出频率，则更新访问时间列表
self.history.pop()
if len(self.history) >= self.num_requests:
# 如果访问时间的频率比 配置中设置的次数多，则返回错误，限制访问
return self.throttle_failure()
return self.throttle_success()
3.基于APIView来写分页功能
1.ListModelMixin的源码分析
(1)ListModelMixin中的list方法
class ListModelMixin:
"""
List a queryset.
"""
def list(self, request, *args, **kwargs):
queryset = self.filter_queryset(self.get_queryset())
# 1 关于分页的部分，通过self.paginate_queryset 产生一个分页类对象，
# self.paginate_queryset就是`GenericAPIView`中的`paginate_queryset`方法
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
(2)GenericAPIView中的paginate_queryset方法
关于分页的部分，通过self.paginate_queryset也就是 GenericAPIView中的paginate_queryset方法产生一个分页类对象，并返回了 分页类对象
def paginate_queryset(self, queryset):
"""
Return a single page of results, or `None` if pagination is disabled.
"""
if self.paginator is None:
return None
return self.paginator.paginate_queryset(queryset, self.request, view=self)
（3）list方法中
返回 分页类对象，将分页类对象传给 【序列化对象】serializer = self.get_serializer(page, many=True)
然后返回一个Response对象 self.get_paginated_response(serializer.data)
（4）PageNumberPagination的get_paginated_response方法
self.get_paginated_response(serializer.data)就是PageNumberPagination的get_paginated_response方法
def get_paginated_response(self, data):
return Response(OrderedDict([
('count', self.page.paginator.count),
('next', self.get_next_link()),
('previous', self.get_previous_link()),
('results', data)
]))
2.代码实现
class BookView(ViewSetMixin, APIView):
def list(self, request):
books = Book.objects.all()
# 1。获取paginator 也就是分页类对象
paginator = CommonPageNumberPagination()
# 2。paginate_queryset(self, queryset, request, view=None)
# queryset就是我们传入的分页的模型对象，view在这里就是self
page = paginator.paginate_queryset(books, request, self)
if page is not None:
# 3 get_serializer(queryset, many=True) 序列化类就是BookSerializer
# 这里的queryset对象指的是分页类对象 page
serializer = BookSerializer(instance=page, many=True)
# return paginator.get_paginated_response(serializer.data)
# 4 PageNumberPagination的get_paginated_response方法
return Response({'count': paginator.page_size,
'next': paginator.get_next_link(),
'previous': paginator.get_previous_link(),
'results': serializer.data})
4.异常处理
1.drf的异常分类
在drf中异常分为两种：(1)drf的异常 (2)非drf的异常
其中，【drf的异常】会被【drf内置的函数】捕获，比如【三大认证，视图类的方法】出现了异常，就会被【内置函数捕获后统一处理】
但是，【非drf的异常】如：程序出错、主动抛出的python异常都不会被处理
所以，在实际工作中无论是否是drf 的异常都应该以统一格式来返回，并记录日志
如
{code:999,'msg':'系统错误，请联系系统管理员'}
DRF的异常处理为：
APIExceptionDjango的异常处理为：
Http404和PermissionDenied
2.异常统一处理
(1)异常处理函数exception_handler
drf中的 处理异常APIException的函数
如果是drf 的异常则返回Request对象，否则返回None
from rest_framework.views import exception_handler
(2)代码演示
excepions.py中
from rest_framework.views import exception_handler
from rest_framework.response import Response
def my_exception_handler(exc, context):
"""
自定义的统一处理drf异常和非drf异常格式的函数
:param exc: 错误对象
:param context: context
:return: 信息字典【响应码，错误信息...】
Response(data, status=exc.status_code, headers=headers)
data是信息字典【响应码，错误信息...】
"""
# 1.通过exception_handler函数捕获异常，由于该函数捕获的不是drf的异常时，返回None
res = exception_handler(exc, context)
# 2. 判断返回值是否是none
if res:
# 3 有值则是drf 的异常，res.data.get('detail')为drf的异常信息
res = Response(data={'code': 111, 'msg': res.data.get('detail', '请联系管理员')})
else:
res = Response(data={'code': 222, 'msg': '非drf错误'})
return res
settings.py中
配置EXCEPTION_HANDLER异常处理函数使用自己编写的
REST_FRAMEWORK = {
'EXCEPTION_HANDLER': 'app01.exceptions.my_exception_handler',
}
- drf的错误
- 非drf的错误
