1 认证
1.1 认证的写法
# 认证的实现
1 写一个类，继承BaseAuthentication，重写authenticate，认证的逻辑写在里面，认证通过，返回两个值(user,auth)，一个值最终给了Requet对象的user，认证失败，抛异常：APIException或者AuthenticationFailed
2 全局使用，局部使用
1.2 认证的源码分析
#1 APIVIew----》dispatch方法---》self.initial(request, *args, **kwargs)---->有认证，权限，频率
#2 只读认证源码： self.perform_authentication(request)
#3 self.perform_authentication(request)就一句话：request.user，需要去drf的Request对象中找user属性（方法）
#4 Request类中的user方法，刚开始来，没有_user,走 self._authenticate()
#5 核心，就是Request类的 _authenticate(self):
def _authenticate(self):
# 遍历拿到一个个认证器，进行认证
# self.authenticators配置的一堆认证类产生的认证类对象组成的 list
#self.authenticators 你在视图类中配置的一个个的认证类：authentication_classes=[认证类1，认证类2]，对象的列表
for authenticator in self.authenticators:
try:
# 认证器(对象)调用认证方法authenticate(认证类对象self, request请求对象)
# 返回值：登陆的用户与认证的信息组成的 tuple
# 该方法被try包裹，代表该方法会抛异常，抛异常就代表认证失败
user_auth_tuple = authenticator.authenticate(self) #注意这self是request对象
except exceptions.APIException:
self._not_authenticated()
raise
# 返回值的处理
if user_auth_tuple is not None:
self._authenticator = authenticator
# 如何有返回值，就将 登陆用户 与 登陆认证 分别保存到 request.user、request.auth
self.user, self.auth = user_auth_tuple
return
# 如果返回值user_auth_tuple为空，代表认证通过，但是没有 登陆用户 与 登陆认证信息，代表游客
self._not_authenticated()
1.3 认证组件的使用
# 写一个认证类 app_auth.py
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from app01.models import UserToken
class MyAuthentication(BaseAuthentication):
def authenticate(self, request):
# 认证逻辑，如果认证通过，返回两个值
#如果认证失败，抛出AuthenticationFailed异常
token=request.GET.get('token')
if token:
user_token=UserToken.objects.filter(token=token).first()
# 认证通过
if user_token:
return user_token.user,token
else:
raise AuthenticationFailed('认证失败')
else:
raise AuthenticationFailed('请求地址中需要携带token')
# 可以有多个认证，从左到右依次执行
# 全局使用，在setting.py中配置
REST_FRAMEWORK={
"DEFAULT_AUTHENTICATION_CLASSES":["app01.app_auth.MyAuthentication",]
}
# 局部使用，在视图类上写
authentication_classes=[MyAuthentication]
# 局部禁用
authentication_classes=[]
2 权限
2.1 权限源码分析
# APIView---->dispatch---->initial--->self.check_permissions(request)(APIView的对象方法)
def check_permissions(self, request):
# 遍历权限对象列表得到一个个权限对象(权限器)，进行权限认证
for permission in self.get_permissions():
# 权限类一定有一个has_permission权限方法，用来做权限认证的
# 参数：权限对象self、请求对象request、视图类对象
# 返回值：有权限返回True，无权限返回False
if not permission.has_permission(request, self):
self.permission_denied(
request, message=getattr(permission, 'message', None)
)
2.2 权限的使用
# 写一个类，继承BasePermission，重写has_permission，如果权限通过，就返回True，不通过就返回False
from rest_framework.permissions import BasePermission
class UserPermission(BasePermission):
def has_permission(self, request, view):
# 不是超级用户，不能访问
# 由于认证已经过了，request内就有user对象了，当前登录用户
user=request.user # 当前登录用户
# 如果该字段用了choice，通过get_字段名_display()就能取出choice后面的中文
print(user.get_user_type_display())
if user.user_type==1:
return True
else:
return False
# 局部使用
class TestView(APIView):
permission_classes = [app_auth.UserPermission]
# 全局使用
REST_FRAMEWORK={
"DEFAULT_AUTHENTICATION_CLASSES":["app01.app_auth.MyAuthentication",],
'DEFAULT_PERMISSION_CLASSES': [
'app01.app_auth.UserPermission',
],
}
# 局部禁用
class TestView(APIView):
permission_classes = []
2.3 内置权限（了解）
# 内置权限的使用：IsAdminUser，控制是否对网站后台有权限的人
# 1 创建超级管理员
# 2 写一个测试视图类
from rest_framework.permissions import IsAdminUser
from rest_framework.authentication import SessionAuthentication
# 演示内置权限，django自带的超级管理员可以查看
# 判断条件request.user.is_staff
class Test3View(APIView):
authentication_classes = [SessionAuthentication, ]
permission_classes = [IsAdminUser, ]
def get(self, request, *args, **kwargs):
return APIResponse(msg='这是Test3View测试数据,超级管理员可以看')
# 3 超级用户登录到admin，再访问test3就有权限
# 4 正常的话，普通管理员，没有权限看（判断的是is_staff字段）
2.4 自定义认证、权限
# 认证+权限
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from rest_framework.permissions import BasePermission
from app01.models import UserToken
class MyAuthentication(BaseAuthentication):
def authenticate(self, request):
# 认证逻辑，如果认证通过，返回两个值
# 如果认证失败，抛出AuthenticationFailed异常
token = request.GET.get('token')
if token:
user_token = UserToken.objects.filter(token=token).first()
# 认证通过
if user_token:
return user_token.user, token
else:
raise AuthenticationFailed('认证失败')
else:
raise AuthenticationFailed('请求地址中未携带token')
class UserPermission(BasePermission):
def has_permission(self, request, view):
# 不是超级用户，不能访问
# 由于认证已经过了，request内就有user对象了，当前登录用户
user = request.user # 当前登录用户
# 如果该字段用了choice，通过get_字段名_display()就能取出choice后面的中文
print(user.get_user_type_display())
if user.user_type == 1:
return True
else:
return False
# models.py
class User(models.Model):
username = models.CharField(max_length=32)
password = models.CharField(max_length=32)
user_type = models.IntegerField(choices=((1, '超级用户'), (2, '普通用户'), (3, '二笔用户')))
class Meta:
verbose_name_plural = '用户表'
class UserToken(models.Model):
token = models.CharField(max_length=64)
user = models.OneToOneField(to='User', on_delete=models.CASCADE) # 一对一关联到User表
class Meta:
verbose_name_plural = '用户token表'
# views.py
class LoginView(APIView):
authentication_classes = []
def post(self, request):
username = request.data.get('username')
password = request.data.get('password')
user = User.objects.filter(username=username, password=password).first()
if user:
# 登陆成功,生成一个随机字符串
token = uuid.uuid4()
# 存到UserToken表中
# models.UserToken.objects.create(token=token,user=user)# 用它每次登陆都会记录一条，不好，如有有记录
# update_or_create有就更新，没有就新增
UserToken.objects.update_or_create(defaults={'token': token}, user=user)
return APIResponse(token=token)
else:
return APIResponse(code=101, msg='用户名或密码错误')
class TestView(APIView):
authentication_classes = [MyAuthentication, ]
permission_classes = [UserPermission, ]
def get(self, request, *args, **kwargs):
return APIResponse(msg='这是TestView测试数据')
# 只要登录用户就可以访问
class Test2View(APIView):
authentication_classes = [MyAuthentication, ]
def get(self, request, *args, **kwargs):
return APIResponse(msg='这是Test2View测试数据')
3 频率
3.1 内置的频率限制(限制未登录用户)
# 全局使用 限制未登录用户1分钟访问5次
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_CLASSES': (
'rest_framework.throttling.AnonRateThrottle',
),
'DEFAULT_THROTTLE_RATES': {
'anon': '3/m',
}
}
# views.py
from rest_framework.permissions import IsAdminUser
from rest_framework.authentication import SessionAuthentication,BasicAuthentication
# 频率限制
# test4 演示全局未登录用户访问频次
class Test4View(APIView):
authentication_classes = []
permission_classes = []
def get(self, request, *args, **kwargs):
return APIResponse(msg='这是Test4View测试数据，登录用户访问频次,全局')
-------------------------------------------------------------------------------------
# 局部使用
# views.pv
from rest_framework.permissions import IsAdminUser
from rest_framework.authentication import SessionAuthentication,BasicAuthentication
from rest_framework.throttling import AnonRateThrottle
# test5 演示局部配置未登录用户访问频次
class Test5View(APIView):
authentication_classes = []
permission_classes = []
throttle_classes = [AnonRateThrottle, ]
def get(self, request, *args, **kwargs):
return APIResponse(msg='这是Test5View测试数据，登录用户访问频次,局部')
3.2 内置频率限制之限制登录用户的访问频次
# 需求：未登录用户1分钟访问5次，登录用户一分钟访问10次
全局：在setting中
'DEFAULT_THROTTLE_CLASSES': (
'rest_framework.throttling.AnonRateThrottle',
'rest_framework.throttling.UserRateThrottle'
),
'DEFAULT_THROTTLE_RATES': {
'user': '10/m',
'anon': '5/m',
}
局部配置：
在视图类中配一个就行
# views.py
# test6 演示登录用户每分钟访问10次，未登录用户访问5次
class Test6View(APIView):
authentication_classes = [SessionAuthentication]
permission_classes = []
throttle_classes = [AnonRateThrottle, UserRateThrottle]
def get(self, request, *args, **kwargs):
return APIResponse(msg='这是Test6View测试数据，录用户每分钟访问10次，未登录用户访问5次')
3.3 根据ip进行频率限制
# 写一个类，继承SimpleRateThrottle，只需要重写get_cache_key
from rest_framework.throttling import ScopedRateThrottle,SimpleRateThrottle
#继承SimpleRateThrottle
class MyThrottle(SimpleRateThrottle):
scope='luffy'
def get_cache_key(self, request, view):
print(request.META.get('REMOTE_ADDR'))
return request.META.get('REMOTE_ADDR') # 返回
# 局部使用
throttle_classes = [MyThrottle,]
# 全局使用
REST_FRAMEWORK={
'DEFAULT_THROTTLE_CLASSES': (
'utils.throttling.MyThrottle',
),
'DEFAULT_THROTTLE_RATES': {
'luffy': '3/m' # key要跟类中的scop对应
},
}
3.4 自定义频率
# 自定制频率类，需要写两个方法
-# 判断是否限次：没有限次可以请求True，限次了不可以请求False
def allow_request(self, request, view):
-# 限次后调用，显示还需等待多长时间才能再访问，返回等待的时间seconds
def wait(self):
# 代码
import time
class IPThrottle():
#定义成类属性,所有对象用的都是这个
VISIT_DIC = {}
def __init__(self):
self.history_list=[]
def allow_request(self, request, view):
'''
#（1）取出访问者ip
#（2）判断当前ip不在访问字典里，添加进去，并且直接返回True,表示第一次访问，在字典里，继续往下走
#（3）循环判断当前ip的列表，有值，并且当前时间减去列表的最后一个时间大于60s，把这种数据pop掉，这样列表中只有60s以内的访问时间，
#（4）判断，当列表小于3，说明一分钟以内访问不足三次，把当前时间插入到列表第一个位置，返回True，顺利通过
#（5）当大于等于3，说明一分钟内访问超过三次，返回False验证失败
'''
ip=request.META.get('REMOTE_ADDR')
ctime=time.time()
if ip not in self.VISIT_DIC:
self.VISIT_DIC[ip]=[ctime,]
return True
self.history_list=self.VISIT_DIC[ip] #当前访问者时间列表拿出来
while True:
if ctime-self.history_list[-1]>60:
self.history_list.pop() # 把最后一个移除
else:
break
if len(self.history_list)<3:
self.history_list.insert(0,ctime)
return True
else:
return False
def wait(self):
# 当前时间，减去列表中最后一个时间
ctime=time.time()
return 60-(ctime-self.history_list[-1])
#全局使用，局部使用
SimpleRateThrottle源码分析
# SimpleRateThrottle源码分析
def get_rate(self):
"""
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 mmmmm
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
# 初次访问缓存为空，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，第一次记录清除，不再算作一次计数
# 10 20 30 40
# 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，代表可以访问
# 将当前时间插入到history列表的开头，将history列表作为数据存到缓存中，key是throttle_user_1，过期时间60s
return self.throttle_success()
4 过滤组件使用
#1 安装：pip3 install django-filter
#2 注册，在app中注册
INSTALLED_APPS = [
'django_filters',
]
#3 全局配，或者局部配
'DEFAULT_FILTER_BACKENDS': ('django_filters.rest_framework.DjangoFilterBackend',)
#4 视图类
class BookView(ListAPIView):
queryset = Book.objects.all()
serializer_class = BookSerializer
filter_fields = ('name',) #配置可以按照哪个字段来过滤
5 排序
# 局部使用和全局使用
# 局部使用
from rest_framework.generics import ListAPIView
from rest_framework.filters import OrderingFilter
from app01.models import Book
from app01.ser import BookSerializer
class Book2View(ListAPIView):
queryset = Book.objects.all()
serializer_class = BookSerializer
filter_backends = [OrderingFilter]
ordering_fields = ('id', 'price')
# urls.py
path('books2/', views.Book2View.as_view()),
]
# 使用：
http://127.0.0.1:8000/books2/?ordering=-price
http://127.0.0.1:8000/books2/?ordering=price
http://127.0.0.1:8000/books2/?ordering=-id