03-Django REST framwork 板块(03-认证、权限、节流)

    xiaoxiao2025-08-27  2

    文章目录

    1. 用户登录认证认证梳理:1. 使用2. 源码流程 2. 权限3.访问频率控制(节流/限流)源码流程梳理

    1. 用户登录认证

    a. 有些API需要用户登录成功之后,才能访问;有些无需登录就能访问。

    b. 基本使用认证组件 解决思路: a. 创建两张表(用户表,token表) b. 用户登录(返回token并保存到数据库)

    c.涉及两种操作(全局认证,单独认证)

    e. 使用框架内置认证类(自己写认证类的时候继承自BaseAuthentication这个类,更规范,不用自己从头搞)

    认证类,必须继承:from rest_framework.authentication import BaseAuthentication 其他认证类:BasicAuthentication(可以浏览器自动生成非模态对话框的一种认证类,SessionAuthentication等等,这些都是基于BaseAuthentication浏览器帮助加密,放到请求头里边传过去得到,基本用不到) 3.大多场景下,继承自BaseAuthentication,然后自己定义认证类 基类如下: class BaseAuthentication(object): """ All authentication classes should extend BaseAuthentication. """ def authenticate(self, request): """ Authenticate the request and return a two-tuple of (user, token). """ raise NotImplementedError(".authenticate() must be overridden.") def authenticate_header(self, request): # 认证失败时,返回的响应头的定义 """ Return a string to be used as the value of the `WWW-Authenticate` header in a `401 Unauthenticated` response, or `None` if the authentication scheme should return `403 Permission Denied` responses. """ pass

    认证梳理:

    1. 使用

    创建类:继承BaseAuthentication; 实现:authenticate方法,其中authenticate_header方法,不用写,直接按基类pass就行

    authenticate的返回值:

    1.返回None,跳过,让下一认证来执行。2.返回(抛出)异常(这里上边类有队异常的接收跟处理)raise exceptions.AuthenticationFailed(用户认证失败 from rest_framework import exceptions)3.返回一个元组(元素1,元素2) # 元素1赋值给request.user; 元素2赋值给request.auth,可以在views.py视图中对返回的两个,进行调用,比如拿到验证通过的对象user(前提定义函数时,得返回user相关的信息)

    局部使用(局部针对某些类进行验证,针对单独的类添加静态字段)

    from rest_framework.authentication import BaseAuthentication,BasicAuthentication class UserInfoView(APIView): """ 订单相关业务 """ authentication_classes = [BasicAuthentication,] # 设置这静态字段,单独设置 def get(self,request,*args,**kwargs): print(request.user) return HttpResponse('用户信息') 全局使用(在settings里边设置配置文件): 全局使用时的文件为路径字符串,如下;而在局部用时,直接添加验证类的名字就行。

    REST_FRAMEWORK = { # 全局使用的认证类

    ```python REST_FRAMEWORK = { # 全局使用的认证类 "DEFAULT_AUTHENTICATION_CLASSES":['api.utils.auth.FirstAuthtication','api.utils.auth.Authtication', ], #里边写的类文件的位置 # "UNAUTHENTICATED_USER":lambda :"匿名用户" "UNAUTHENTICATED_USER":None, # 匿名,request.user = None "UNAUTHENTICATED_TOKEN":None,# 匿名,request.auth = None }

    2. 源码流程

    dispatch 封装request 获取定义的认证类(全局/局部),通过列表生成时创建对象。 initial perform_authentication request.user(内部循环…) - user - _authenticate(定义返回内容,有返回元组,没有,返回个元组)

    2. 权限

    class MyPermission(object): def has_permission(self,request,view): if request.user.user_type != 3: return False return True class OrderView(APIView): """ 订单相关业务(只有SVIP用户有权限) """ permission_classes = [MyPermission,] def get(self,request,*args,**kwargs): # request.user # request.auth self.dispatch ret = {'code':1000,'msg':None,'data':None} try: ret['data'] = ORDER_DICT except Exception as e: pass return JsonResponse(ret)

    权限知识梳理: 同上,为了代码规范,最好也继承自BasePermission

    类,必须继承:BasePermission,必须实现:has_permission方法,其中还有个has_object_persion方法,用来细化到针对对象的权限判断 from rest_framework.permissions import BasePermission class SVIPPermission(BasePermission): message = "必须是SVIP才能访问" def has_permission(self,request,view): if request.user.user_type != 3: return False return True 返回值: True, 有权访问False,无权访问 局部使用 class UserInfoView(APIView): """ 订单相关业务(普通用户、VIP) """ permission_classes = [MyPermission1, ] def get(self,request,*args,**kwargs): return HttpResponse('用户信息') 全局使用 REST_FRAMEWORK = { "DEFAULT_PERMISSION_CLASSES":['api.utils.permission.SVIPPermission'] }

    3.访问频率控制(节流/限流)

    控制请求的访问频率,具体实现

    代码实现是拿ip来记录,对于没登录的匿名用户,我们能拿到的也只有ip,要是换ip访问,这个控制不了

    访问的ip地址的获得方式 : request.META.get('REMOTE_ADDR)

    1.类, 若继承:BaseThrottle,要自己实现:allow_request、wait2.类, 若继承:SimpleRateThrottle,只需实现实现:get_cache_key、其中的一个字段:scope = “Luffy”(配置文件中限制节流频率的key) import time VISIT_RECORD = {} class VisitThrottle(object): """60s内只能访问3次""" def __init__(self): self.history = None def allow_request(self,request,view): # 1. 获取用户IP remote_addr = request.META.get('REMOTE_ADDR') ctime = time.time() if remote_addr not in VISIT_RECORD: VISIT_RECORD[remote_addr] = [ctime,] return True history = VISIT_RECORD.get(remote_addr) self.history = history while history and history[-1] < ctime - 60: history.pop() if len(history) < 3: history.insert(0,ctime) return True # return True # 表示可以继续访问 # return False # 表示访问频率太高,被限制 def wait(self): """ 还需要等多少秒才能访问 :return: """ ctime = time.time() return 60 - (ctime - self.history[-1]) class AuthView(APIView): """ 用于用户登录认证 """ authentication_classes = [] permission_classes = [] throttle_classes = [VisitThrottle,] def post(self,request,*args,**kwargs): ret = {'code':1000,'msg':None} try: user = request._request.POST.get('username') pwd = request._request.POST.get('password') obj = models.UserInfo.objects.filter(username=user,password=pwd).first() if not obj: ret['code'] = 1001 ret['msg'] = "用户名或密码错误" # 为登录用户创建token token = md5(user) # 存在就更新,不存在就创建 models.UserToken.objects.update_or_create(user=obj,defaults={'token':token}) ret['token'] = token except Exception as e: ret['code'] = 1002 ret['msg'] = '请求异常' return JsonResponse(ret)

    源码流程梳理

    from rest_framework.throttling import BaseThrottle, SimpleRateThrottle # 代码中的get_ident()获得标识,不仅可以用远程IP,或者自己写的标识,也可以获得传输过来的代理的Ip,但是前提是请求中必须包含过来的ip数据,不然也是无法实现的 class MyThrottle(SimpleRateThrottle): """限制所有用户""" scope = 'Luffy' def get_cache_key(self, request, view): #设置的进行频率限制时的唯一标识。缓存的键,可以设置远程ip为标识,利用get_ident() return self.get_ident(request) class UserThrottle(SimpleRateThrottle): """限制已经登录用户的访问频率""" scope = 'LuffyUser' def get_cache_key(self, request, view): #设置进行频率限制的唯一标识。缓存的键,可设置远程ip为标识,利用get_ident() return request.user.username 全局使用 REST_FRAMEWORK = { REST_FRAMEWORK = { # 全局使用的认证类 "DEFAULT_AUTHENTICATION_CLASSES":['api.utils.auth.FirstAuthtication','api.utils.auth.Authtication', ], # "DEFAULT_AUTHENTICATION_CLASSES":['api.utils.auth.FirstAuthtication', ], # "UNAUTHENTICATED_USER":lambda :"匿名用户" "UNAUTHENTICATED_USER":None, # 匿名,request.user = None "UNAUTHENTICATED_TOKEN":None,# 匿名,request.auth = None "DEFAULT_PERMISSION_CLASSES":['api.utils.permission.SVIPPermission'], # 这里设置的是针对所有用户频率的限制,针对单独的为登录用户的设置可以直接在auth认证那里 # 直接添加字段;如果频率一样的话,可以直接针对ip进行全局的限制。 "DEFAULT_THROTTLE_CLASSES":["api.utils.throttle.UserThrottle"], "DEFAULT_THROTTLE_RATES":{ "Luffy":'3/m', # 3次每分钟(SimpleRateThrottle内部方法实现) "LuffyUser":'10/m', # } } 局部单独使用,直接加导视图函数内就行throttle_classes = [UserThrottle,](见上方代码)
    最新回复(0)