一. 认证组件
1. 流程
1. 写一个类,继承BaseAuthentication,重写authenticate,认证的逻辑写在里面. 认证通过,返回两个值,一个值最终给了包装以后的request对象, 视图中就可以通过request.user获取, 认证失败,抛异常:APIException 或者 AuthenticationFailed 注意: 本质抛出的异常只是AuthenticationFailed, 而AuthenticationFailed又继承了APIException, 因此源码中只需要捕获父类异常, 在属性查找时必然会找到其子类AuthenticationFailed class AuthenticationFailed(APIException): status_code = status.HTTP_401_UNAUTHORIZED default_detail = _('Incorrect authentication credentials.') default_code = 'authentication_failed' 提示: BaseAuthentication可以不继承, BaseAuthentication作用仅仅是对继承它的子类必须要定义authentication方法 本质采用的就是通过抛出异常的形式去规范认证的方法名写法. def authenticate(self, request): raise NotImplementedError(".authenticate() must be overridden.") 2. 全局使用认证,局部使用认证
2. 认证的源码分析
APIView--->dispatch方法--->self.initial(request, *args, **kwargs)--->认证、权限、频率 self.perform_authentication(request) # 只读认证--->request.user,需要去drf的Request对象中user属性(方法) Request类中的user方法,刚开始,没有_user,走self_authticate() dispatch方法--->request = self.initialize_request(request, *args, **kwargs)---> authenticators=self.get_authenticators()--->列表生成式 return [auth() for auth in self.authentication_classes]--->在request被Request实例化的时候,通过列表生成式,把认证类实例化成认证对象,并列表对象给Request类中的__init__方法中的self.authenticators = authenticators or () # 这是核心 def _authenticate(self): # 遍历拿到一个认证器,进行认证 # self.authenticators是在视图类中配置的一个个的认证类对象的 列表 for authenticator in self.authenticators: try: # 认证对象self,是request请求对象 # 返回值:登录的用户与认证的信息组成的tuple user_auth_tuple = authenticator.authenticate(self) # authenticator是对象,self是requset对象,传进来的,这句话就认证对象执行方法,就去认证类中找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 # 结束循环,后面的认证不在执行 # user_auth_tuple为空,代表认证通过,但是没有,表是游客 self._not_authenticated()
3. 自定义认证功能实现
局部使用
app01_auth.py
''' @作者:xiaobao @联系方式:[email protected] ''' from rest_framework.authentication import BaseAuthentication from rest_framework.exceptions import AuthenticationFailed from app01 import models class AuthLoginView(BaseAuthentication): def authenticate(self, request): # 认证逻辑,如果认证通过,返回两个值(源码,返回了一个元祖,所以要返回两个值 # 如果认证失败,抛出AuthenticationFailed异常 token = request.GET.get('token') if token: user_token = models.UserToken.objects.filter(token=token).first() # 认证通过 if user_token: return user_token.userinfo, token else: raise AuthenticationFailed('认证未通过') else: raise AuthenticationFailed('请求没有token')
views.py
视图类中加认证类列表
from app01.app01_auth import AuthLoginView class BookViewSet(ModelViewSet): # 认证 authentication_classes = [AuthLoginView] queryset = models.Book.objects.all() serializer_class = BookSerializer from rest_framework.views import APIView from rest_framework.response import Response from app01 import models import uuid class LoginView(APIView): def post(self, request): username = request.data.get('username') password = request.data.get('password') user_obj = models.UserInfo.objects.filter(username=username, password=password).first() if user_obj: # 登录成功,生成一个随机字符串 token = uuid.uuid4() # 每次登录一次都会记录一条,不好 # models.UserToken.objects.create(token=token, userinfo=user_obj) # user_obj有值,就存token,没有,就不存 models.UserToken.objects.update_or_create(defaults={'token': token}, userinfo=user_obj) return Response({'status': 100, 'msg': '登录成功', 'token': token}) else: return Response({'status': 101, 'msg': '用户名或密码错误'})
全局使用
在settings.py中设置
REST_FRAMEWORK = { "DEFAULT_AUTHENTICATION_CLASSES": ["app01.app01_auth.AuthLoginView", ] # 里面是路径字符串 } # 局部禁用,视图类中加 authentication_classes = []
二. 权限组件
1. 自定义权限
1 权限源码
# APIView--->dispatch--->initial--->self.check_permissions(request) def check_permissions(self, request): """ Check if the request should be permitted. Raises an appropriate exception if the request is not permitted. """ # 遍历权限对象列表得到一个个权限对象(权限器),进行权限认证 for permission in self.get_permissions(): # 权限类一定有一个has_permission方法,用来做权限认证的 # 参数:权限对象self、请求对象request、视图类对象 # 返回值:有权限返回True,无权限返回False if not permission.has_permission(request, self): # 对象点has_permission方法,里面传参了两个参数,权限类中的该方法,需要传三个参数 self.permission_denied( request, message=getattr(permission, 'message', None), code=getattr(permission, 'code', None) )
2 权限的使用
app01_auth.py
from rest_framework.permissions import BasePermission class UserPermission(BasePermission): def has_permission(self, request, view): # view是视图类传给来的对象 # 不是超级用户,不能访问 # 由于认证已经过了,request内就有了user对象了,当前登录用户 user = request.user print(user.get_user_type_display()) # 显示choice字段,数字对应的值,写法是get_字段名_display if user.user_type == 1: return True else: return False
views.py
# 使用方法和认证一样,分全局配置和局部配置。局部配置那个视图类需要超级管理员,就配置permission_classes= [UserPermission],局部配置,就是在setting配置了,不那个视图类有超级管理权限,就配置在那个视图类中配置permission_classes= []。 class TestView(APIView): # 局部配置,超级用户才能看 permission_classes = [UserPermission] def get(self, request): return Response('超级用户') class TestView1(APIView): def get(self, request): return Response('普通用户')
总结
1. 新建.py文件书写权限控制类, 继承 BasePermission from rest_framework.permissions import BasePermission 2. 新建的类中重写 has_permission 方法 三个参数: self, request, view self: 自定义认证类的丢下 request: APIView中的dispatch中封装过后的request对象 view: 自定义视图类实例化过后的对象 3. 根据has_permission的返回布尔值是否是True 或者 False进行判断 True: 权限通过 False: 权限不通过, 抛出异常(抛出2种不同类型异常, 根据实际情况分析) 4. 全局配置 或者 局部配置
2. 内置权限类
内置权限类
from rest_framework.permissions import AllowAny,IsAuthenticated,IsAdminUser,IsAuthenticatedOrReadOnly - AllowAny 允许所有用户 - IsAuthenticated 仅通过认证的用户 - IsAdminUser 仅管理员用户 - IsAuthenticatedOrReadOnly 已经登陆认证的用户可以对数据进行增删改操作,没有登陆认证的只能查看数据。
全局配置 和 局部配置
# 全局配置 REST_FRAMEWORK = { 'DEFAULT_PERMISSION_CLASSES': [ rest_framework.permissions.AllowAny', # 内置 ], } # 局部配置 ''' # IsAdminUser return bool(request.user and request.user.is_staff) # SessionAuthentication user = getattr(request._request, 'user', None) if not user or not user.is_active: return None SessionAuthentication源码控制的是user 和 user.is_active 控制情况: 匿名用户user.is_action布尔值是False 1. 原生的request对象中时候有user属性 2. user中is_active的布尔值为True. ''' from rest_framework.authentication import SessionAuthentication from rest_framework.permissions import IsAdminUser authentication_classes = [SessionAuthentication] permission_classes = [IsAdminUser]
代码实例
# 内置局部认证+内置局部权限控制: is_active控制认证 is_staff控制权限 from rest_framework.response import Response from rest_framework.views import APIView from rest_framework.authentication import SessionAuthentication from rest_framework.permissions import IsAdminUser class TextView2(APIView): """ SessionAuthentication认证的判断依据: is_active if not user or not user.is_active: return None 失败返回: Authentication credentials were not provided. IsAdminUser权限的判断依据: is_staff return bool(request.user and request.user.is_staff) 失败返回: You do not have permission to perform this action. """ authentication_classes = [SessionAuthentication] permission_classes = [IsAdminUser] def get(self, request): return Response('这是活跃的工作
总结
内置认证: from rest_framework.authentication import SessionAuthentication 控制的是is_active 内置权限: from rest_framework.permissions import isAdminUser 控制的是is_staff
三 频率限制
1. 自定义频率类
1) 编写频率类
# 自定义的逻辑 #(1)取出访问者ip #(2)判断当前ip不在访问字典里,添加进去,并且直接返回True,表示第一次访问,在字典里,继续往下走 #(3)循环判断当前ip的列表,有值,并且当前时间减去列表的最后一个时间大于60s,把这种数据pop掉,这样列表中只有60s以内的访问时间, #(4)判断,当列表小于3,说明一分钟以内访问不足三次,把当前时间插入到列表第一个位置,返回True,顺利通过 #(5)当大于等于3,说明一分钟内访问超过三次,返回False验证失败 class MyThrottles(): VISIT_RECORD = {} def __init__(self): self.history=None def allow_request(self,request, view): #(1)取出访问者ip # print(request.META) ip=request.META.get('REMOTE_ADDR') import time ctime=time.time() # (2)判断当前ip不在访问字典里,添加进去,并且直接返回True,表示第一次访问 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,顺利通过 # (5)当大于等于3,说明一分钟内访问超过三次,返回False验证失败 if len(self.history)<3: self.history.insert(0,ctime) return True else: return False def wait(self): import time ctime=time.time() return 60-(ctime-self.history[-1])
2) 全局配置
REST_FRAMEWORK = { 'DEFAULT_THROTTLE_CLASSES':['app01.utils.MyThrottles',], }
3) 局部配置
#在视图类里使用 throttle_classes = [MyThrottles,]
2. 内置频率
可以对接口访问的频次进行限制,以减轻服务器压力。
一般用于付费购买次数,投票等场景使用
可以在配置文件中,使用DEFAULT_THROTTLE_CLASSES
和 DEFAULT_THROTTLE_RATES
进行全局配置
REST_FRAMEWORK = { 'DEFAULT_THROTTLE_CLASSES': ( 'rest_framework.throttling.AnonRateThrottle', 'rest_framework.throttling.UserRateThrottle' ), 'DEFAULT_THROTTLE_RATES': { 'anon': '100/day',# m分,h时,s秒 'user': '1000/day' } }
DEFAULT_THROTTLE_RATES 可以使用 second, minute, hour 或day来指明周期。
也可以在具体视图中通过throttle_classess属性来配置,如
from rest_framework.throttling import UserRateThrottle from rest_framework.views import APIView class ExampleView(APIView): throttle_classes = (UserRateThrottle,)
限流类型
1) AnonRateThrottle
限制所有匿名未认证用户,使用IP区分用户。
使用DEFAULT_THROTTLE_RATES['anon']
来设置频次
2)UserRateThrottle
限制认证用户,使用User id 来区分。
使用DEFAULT_THROTTLE_RATES['user']
来设置频次
3)ScopedRateThrottle
限制用户对于每个视图的访问频次,使用ip或user id。
class ContactListView(APIView): throttle_scope = 'contacts' ... class ContactDetailView(APIView): throttle_scope = 'contacts' ... class UploadView(APIView): throttle_scope = 'uploads' ... REST_FRAMEWORK = { 'DEFAULT_THROTTLE_CLASSES': ( 'rest_framework.throttling.ScopedRateThrottle', ), 'DEFAULT_THROTTLE_RATES': { 'contacts': '1000/day', 'uploads': '20/day' } }
登录用户的频率限制,如果是用的内置频率限制,认证是用的内置权限,如果是自定义认证权限,就需要用自定义内置频率。
标签:self,request,认证,framework,user,import,权限,drf From: https://www.cnblogs.com/coderxueshan/p/17730915.html