为了方便接下来的学习,我们创建一个新的子应用 opt
1 python manage.py startapp opt
因为接下来的功能中需要使用到登陆功能,所以我们使用django内置admin站点并创建一个管理员.
1 2 3 python3 manage.py makemigrations python3 manage.py migrate python3 manage.py createsuperuser
创建管理员以后,访问admin站点,先修改站点的语言配置
settings.py
1 2 3 4 5 6 7 8 9 LANGUAGE_CODE = 'zh-hans' TIME_ZONE = 'Asia/Shanghai' USE_I18N = True USE_L10N = True USE_TZ = True
访问admin 站点效果:
一 认证Authentication 1.1 自定义认证方案 1.1.1 编写models 1 2 3 4 5 6 7 8 9 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 UserToken (models.Model): user=models.OneToOneField(to='User' ) token=models.CharField(max_length=64 )
1.1.2 新建认证类 1 2 3 4 5 6 7 8 9 10 11 from rest_framework.authentication import BaseAuthenticationclass TokenAuth (): def authenticate (self, request ): token = request.GET.get('token' ) token_obj = models.UserToken.objects.filter (token=token).first() if token_obj: return else : raise AuthenticationFailed('认证失败' ) def authenticate_header (self,request ): pass
1.1.3 编写视图 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 def get_random (name ): import hashlib import time md=hashlib.md5() md.update(bytes (str (time.time()),encoding='utf-8' )) md.update(bytes (name,encoding='utf-8' )) return md.hexdigest() class Login (APIView ): def post (self,reuquest ): back_msg={'status' :1001 ,'msg' :None } try : name=reuquest.data.get('name' ) pwd=reuquest.data.get('pwd' ) user=models.User.objects.filter (username=name,password=pwd).first() if user: token=get_random(name) models.UserToken.objects.update_or_create(user=user,defaults={'token' :token}) back_msg['status' ]='1000' back_msg['msg' ]='登录成功' back_msg['token' ]=token else : back_msg['msg' ] = '用户名或密码错误' except Exception as e: back_msg['msg' ]=str (e) return Response(back_msg) class Course (APIView ): authentication_classes = [TokenAuth, ] def get (self, request ): return HttpResponse('get' ) def post (self, request ): return HttpResponse('post' )
1.1.4 全局使用 1 2 3 REST_FRAMEWORK={ "DEFAULT_AUTHENTICATION_CLASSES" :["app01.service.auth.Authentication" ,] }
1.1.5 局部使用 1 2 authentication_classes = [TokenAuth, ]
1.2 内置认证方案(需要配合权限使用) 可以在配置文件中配置全局默认的认证方案
1 2 3 4 5 6 REST_FRAMEWORK = { 'DEFAULT_AUTHENTICATION_CLASSES' : ( 'rest_framework.authentication.SessionAuthentication' , 'rest_framework.authentication.BasicAuthentication' , ) }
也可以在每个视图中通过设置authentication_classess属性来设置
1 2 3 4 5 6 7 from rest_framework.authentication import SessionAuthentication, BasicAuthenticationfrom rest_framework.views import APIViewclass ExampleView (APIView ): authentication_classes = [SessionAuthentication, BasicAuthentication] ...
认证失败会有两种可能的返回值:
401 Unauthorized 未认证
403 Permission Denied 权限被禁止
1.3认证类源码分析 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 APIView的initial方法----》有三句话 self.perform_authentication(request) self.check_permissions(request) self.check_throttles(request) def _authenticate (self ): for authenticator in self.authenticators: try : user_auth_tuple = authenticator.authenticate(self) 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 return Request( request, parsers=self.get_parsers(), authenticators=self.get_authenticators(), negotiator=self.get_content_negotiator(), parser_context=parser_context ) def get_authenticators (self ): return [auth() for auth in self.authentication_classes] return [LoginAuth(),]
二 权限Permissions 权限控制可以限制用户对于视图的访问和对于具体数据对象的访问。
在执行视图的dispatch()方法前,会先进行视图访问权限的判断
在通过get_object()获取具体对象时,会进行模型对象访问权限的判断
2.1 自定义权限 2.1.1 编写权限类 1 2 3 4 5 6 7 8 9 10 11 12 13 14 from rest_framework.permissions import BasePermissionclass UserPermission (BasePermission ): message = '不是超级用户,查看不了' def has_permission (self, request, view ): user_type = request.user.user_type print (user_type) if user_type == 1 : return True else : return False
2.1.2 全局使用 1 2 3 4 REST_FRAMEWORK={ "DEFAULT_AUTHENTICATION_CLASSES" :["app01.service.auth.Authentication" ,], "DEFAULT_PERMISSION_CLASSES" :["app01.service.permissions.SVIPPermission" ,] }
2.1.3 局部使用 1 2 permission_classes = [UserPermission,]
2.1.4 说明 1 2 3 4 5 6 如需自定义权限,需继承rest_framework.permissions.BasePermission父类,并实现以下两个任何一个方法或全部 - `.has_permission(self, request, view)` 是否可以访问视图, view表示当前视图对象 - `.has_object_permission(self, request, view, obj)` 是否可以访问数据对象, view表示当前视图, obj为数据对象
2.2 内置权限 2.2.1 内置权限类 1 2 3 4 5 from rest_framework.permissions import AllowAny,IsAuthenticated,IsAdminUser,IsAuthenticatedOrReadOnly- AllowAny 允许所有用户 - IsAuthenticated 仅通过认证的用户 - IsAdminUser 仅管理员用户 - IsAuthenticatedOrReadOnly 已经登陆认证的用户可以对数据进行增删改操作,没有登陆认证的只能查看数据。
2.2.2 全局使用 可以在配置文件中全局设置默认的权限管理类,如
1 2 3 4 5 6 7 REST_FRAMEWORK = { .... 'DEFAULT_PERMISSION_CLASSES' : ( 'rest_framework.permissions.IsAuthenticated' , ) }
如果未指明,则采用如下默认配置
1 2 3 'DEFAULT_PERMISSION_CLASSES' : ( 'rest_framework.permissions.AllowAny' , )
2.2.3 局部使用 也可以在具体的视图中通过permission_classes属性来设置,如
1 2 3 4 5 6 from rest_framework.permissions import IsAuthenticatedfrom rest_framework.views import APIViewclass ExampleView (APIView ): permission_classes = (IsAuthenticated,) ...
2.2.4 实际操作 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 'DEFAULT_PERMISSION_CLASSES' : ( 'rest_framework.permissions.IsAuthenticated' , ) path('test/' , views.TestView.as_view()), class TestView (APIView ): def get (self,request ): return Response({'msg' :'个人中心' }) rest_framework.permissions.IsAdminUser
三 限流Throttling 可以对接口访问的频次进行限制,以减轻服务器压力。
一般用于付费购买次数,投票等场景使用.
3.1 自定义频率类 3.1.1 编写频率类 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 class MyThrottles (): VISIT_RECORD = {} def __init__ (self ): self.history=None def allow_request (self,request, view ): ip=request.META.get('REMOTE_ADDR' ) import time ctime=time.time() if ip not in self.VISIT_RECORD: self.VISIT_RECORD[ip]=[ctime,] return True self.history=self.VISIT_RECORD.get(ip) while self.history and ctime-self.history[-1 ]>60 : self.history.pop() 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 ])
3.1.2 全局使用 1 2 3 REST_FRAMEWORK = { 'DEFAULT_THROTTLE_CLASSES' :['app01.utils.MyThrottles' ,], }
3.1.3 局部使用 1 2 throttle_classes = [MyThrottles,]
3.2 内置频率类 3.2.1 根据用户ip限制 1 2 3 4 5 6 7 8 9 10 11 12 13 14 from rest_framework.throttling import SimpleRateThrottleclass VisitThrottle (SimpleRateThrottle ): scope = 'luffy' def get_cache_key (self, request, view ): return self.get_ident(request) REST_FRAMEWORK = { 'DEFAULT_THROTTLE_RATES' :{ 'luffy' :'3/m' } }
了解:错误信息中文显示
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 class Course (APIView ): authentication_classes = [TokenAuth, ] permission_classes = [UserPermission, ] throttle_classes = [MyThrottles,] def get (self, request ): return HttpResponse('get' ) def post (self, request ): return HttpResponse('post' ) def throttled (self, request, wait ): from rest_framework.exceptions import Throttled class MyThrottled (Throttled ): default_detail = '傻逼啊' extra_detail_singular = '还有 {wait} second.' extra_detail_plural = '出了 {wait} seconds.' raise MyThrottled(wait)
3.2.2 限制匿名用户每分钟访问3次 1 2 3 4 5 6 7 8 9 10 REST_FRAMEWORK = { 'DEFAULT_THROTTLE_CLASSES' : ( 'rest_framework.throttling.AnonRateThrottle' , ), 'DEFAULT_THROTTLE_RATES' : { 'anon' : '3/m' , } }
3.2.3 限制登陆用户每分钟访问10次 1 2 3 4 5 6 7 8 9 REST_FRAMEWORK = { 'DEFAULT_THROTTLE_CLASSES' : ( 'rest_framework.throttling.UserRateThrottle' ), 'DEFAULT_THROTTLE_RATES' : { 'user' : '10/m' } }
3.2.4 其他 1) AnonRateThrottle
限制所有匿名未认证用户,使用IP区分用户。
使用DEFAULT_THROTTLE_RATES['anon']
来设置频次
2)UserRateThrottle
限制认证用户,使用User id 来区分。
使用DEFAULT_THROTTLE_RATES['user']
来设置频次
3)ScopedRateThrottle
限制用户对于每个视图的访问频次,使用ip或user id。
例如:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 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' } }
3.3 SimpleRateThrottle源码 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 APIview中执行check_throttles ---> 拿到自定义视图类中的throttle_classes(SimpleRateThrottle),执行 SimpleRateThrottle的allow_request方法,判断是否频率超限制,返回True or False ---> 如果 allow_request 返回 False ,表示访问评率超限制,则返回还有多长时间可以重新访问 def check_throttles (self, request ): throttle_durations = [] for throttle in self.get_throttles(): if not throttle.allow_request(request, self): throttle_durations.append(throttle.wait()) if throttle_durations: durations = [ duration for duration in throttle_durations if duration is not None ] duration = max (durations, default=None ) self.throttled(request, duration) def allow_request (self, request, view ): 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() 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()
实例 全局配置中设置访问频率
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 'DEFAULT_THROTTLE_RATES' : { 'anon' : '3/minute' , 'user' : '10/minute' } from rest_framework.authentication import SessionAuthenticationfrom rest_framework.permissions import IsAuthenticatedfrom rest_framework.generics import RetrieveAPIViewfrom rest_framework.throttling import UserRateThrottleclass StudentAPIView (RetrieveAPIView ): queryset = Student.objects.all () serializer_class = StudentSerializer authentication_classes = [SessionAuthentication] permission_classes = [IsAuthenticated] throttle_classes = (UserRateThrottle,)