django学习笔记--drf认证、权限、限流记录

drf基础配置

版本：

Django              3.2
djangorestframework 3.13.1

纯净版配置：

INSTALLED_APPS = [# 'django.contrib.admin',# 'django.contrib.auth',# 'django.contrib.contenttypes',# 'django.contrib.sessions',# 'django.contrib.messages','django.contrib.staticfiles','api.apps.ApiConfig','rest_framework'
]MIDDLEWARE = ['django.middleware.security.SecurityMiddleware',# 'django.contrib.sessions.middleware.SessionMiddleware','django.middleware.common.CommonMiddleware',# 'django.middleware.csrf.CsrfViewMiddleware',# 'django.contrib.auth.middleware.AuthenticationMiddleware',# 'django.contrib.messages.middleware.MessageMiddleware','django.middleware.clickjacking.XFrameOptionsMiddleware',
]TEMPLATES = [{'BACKEND': 'django.template.backends.django.DjangoTemplates','DIRS': [],'APP_DIRS': True,'OPTIONS': {'context_processors': ['django.template.context_processors.debug','django.template.context_processors.request',# 'django.contrib.auth.context_processors.auth',# 'django.contrib.messages.context_processors.messages',],},},
]REST_FRAMEWORK = {"UNAUTHENTICATED_USER": None
}

路由

from django.contrib import admin
from django.urls import path
from api import viewsurlpatterns = [# path('admin/', admin.site.urls),path('users/', views.UserView.as_view()),
]```
###视图
```python
from rest_framework.views import APIView
from rest_framework.response import Responseclass UserView(APIView):def get(self, request):return Response("...")

request参数：

路径参数

# urls.py
urlpatterns = [path('users/<str:version>/<int:pid>/',views.UserView.as_view())
]
# views.py
class UserView(APIView):def get(self, request,version,pid):print(version,pid)print(self.kwargs) #{version:'v1','pid':123}return Response("ok")

drf request对象获取参数示例：

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticatedclass MyView(APIView):permission_classes = [IsAuthenticated]def post(self, request, *args, **kwargs):# 获取路径参数item_id = kwargs.get('item_id')# 获取查询参数search_query = request.query_params.get('search', '')# 获取请求头auth_token = request.headers.get('Authorization')# 获取表单或 JSON 数据data_field = request.data.get('data_field')# 获取文件uploaded_file = request.FILES.get('file')# 获取用户信息user = request.userreturn Response({"item_id": item_id,"search_query": search_query,"auth_token": auth_token,"data_field": data_field,"uploaded_file_name": uploaded_file.name if uploaded_file else None,"user": {"username": user.username, "email": user.email},})

认证

#基础类，定义一个认证类，需要实现如下基础类两个方法
class BaseAuthentication:def authenticate(self, request):raise NotImplementedError(".authenticate() must be overridden.")def authenticate_header(self, request):pass
# 简单实现示例：
from rest_framework.authentication import BaseAuthentication
from api.models import Userclass MyAuthentication(BaseAuthentication):def authenticate(self, request):token=request.headers.get('token')user=User.objects.filter(token=token).first()if user:return user,tokenreturn None,Nonedef authenticate_header(self, request):return 'Token'

源码路径

APIView,dispatch方法会调用self.initial(request, *args, **kwargs)
self.initial会依次调用下面三个方法分别为认证、权限、限流

        self.perform_authentication(request)self.check_permissions(request)self.check_throttles(request)

下面我们先看下认证源码，上述的self都是APIView的子类对象， self.perform_authentication(request)调用request.user进入Request对象，调用Request如下方法，self.authenticators是配置在settings.REST_FRAMEWORK下的DEFAULT_AUTHENTICATION_CLASSES属性或者APIView子类的authentication_classes属性配置里，优先取view类里配置的，没有再取settings公共的

    def _authenticate(self):for authenticator in self.authenticators:try:user_auth_tuple = authenticator.authenticate(self)except exceptions.APIException:self._not_authenticated()raiseif user_auth_tuple is not None:self._authenticator = authenticatorself.user, self.auth = user_auth_tuplereturnself._not_authenticated()

备注：认证类默认前一个未认证会继续下一个，有一个认证成功就会终止认证继续后面逻辑，若认证抛出异常也会终止认证并返回，所有认证都没通过也会继续执行后面逻辑

权限

#自定义权限类，限制某个角色才能访问
from rest_framework.permissions import BasePermissionclass PermissionA(BasePermission):message = {"code": 1003, 'data': "无权访问"}def has_permission(self, request, view):if request.user.role == 2:return Truereturn False# 暂时先这么写def has_object_permission(self, request, view, obj):return True

源码简单分析:读取当前View类（如：UserView）配置的permission_classes，若未配置则取APIView默认配置settings.REST_FRAMEWORK的DEFAULT_PERMISSION_CLASSES属性

    def check_permissions(self, request):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))

备注：权限类默认有一个失败就返回异常，若想定制满足任意权限，可在APIView子类修改check_permission方法

def check_permissions(self, request):for permission in self.get_permissions():if permission.has_permission(request, self):return #有任一权限返回#无任何权限通过走此逻辑self.permission_denied(request,message=getattr(permission, 'message', None),code=getattr(permission, 'code', None))

限流

class MyRateThrottle(SimpleRateThrottle):cache = default_cache  # 访问记录存放在django的缓存中（需设置缓存）scope = "user"  # 构造缓存中的keycache_format = 'throttle_%(scope)s_%(ident)s''''设置访问频率，例如：1分钟允许访问10次,可在settings全局定义REST_FRAMEWORK = {"DEFAULT_THROTTLE_CLASSES":["xxx.xxx.xx.限流类", ],"DEFAULT_THROTTLE_RATES": {"user": "10/m",  #与scope对应，没有值会抛出异常"xx":"100/h"}}'''THROTTLE_RATES = {"user": "10/s"}#没有定义会去全局定义里取def get_cache_key(self, request, view):if request.user:ident = request.user.pk  # 用户IDelse:ident = self.get_ident(request)  # 获取请求用户IP（去request中找请求头）#throttle_user_11.11.11.11return self.cache_format % {'scope': self.scope, 'ident': ident}def throttle_failure(self):wait = self.wait()detail = {"code": 1005,"data": "访问频率限制",'detail': "需等待{}s才能访问".format(int(wait))}raise ThrottledException(detail)

使用限流需要使用django-redis，安装pip install django-redis

# cache缓存配置
CACHES = {"default": {"BACKEND": "django_redis.cache.RedisCache","LOCATION": "redis://192.168.31.125:6379/0","OPTIONS": {"CLIENT_CLASS": "django_redis.client.DefaultClient","CONNECTION_POOL_KWARGS": {"max_connections": 100},'decode_responses': True,"PASSWORD": "xxxx",},}
}