首页 > 其他分享 >django学习笔记--drf认证、权限、限流记录

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

时间:2025-01-05 22:44:28浏览次数:1  
标签:permission -- self request django 限流 user def

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 views

urlpatterns = [
    # path('admin/', admin.site.urls),
    path('users/', views.UserView.as_view()),
]```
###视图
```python
from rest_framework.views import APIView
from rest_framework.response import Response


class 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 IsAuthenticated

class 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.user

        return 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 User

class MyAuthentication(BaseAuthentication):

    def authenticate(self, request):
        token=request.headers.get('token')
        user=User.objects.filter(token=token).first()
        if user:
            return user,token
        return None,None

    def 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()
                raise

            if user_auth_tuple is not None:
                self._authenticator = authenticator
                self.user, self.auth = user_auth_tuple
                return

        self._not_authenticated()

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

权限

#自定义权限类,限制某个角色才能访问
from rest_framework.permissions import BasePermission

class PermissionA(BasePermission):
    message = {"code": 1003, 'data': "无权访问"}

    def has_permission(self, request, view):
        if request.user.role == 2:
            return True
        return 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"  # 构造缓存中的key
    cache_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  # 用户ID
        else:
            ident = self.get_ident(request)  # 获取请求用户IP(去request中找请求头)

        #throttle_user_11.11.11.11

        return 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",
        },
    }
}

标签:permission,--,self,request,django,限流,user,def
From: https://www.cnblogs.com/liuzhiyangz/p/18653853

相关文章

  • 一个运行时打桩的模板
    被打桩的函数是STUB_FUN,可以替换成如__cudaRegisterFunction,另外插入的函数zwl_profiler可另外定义, .text .section .rodata.LC0: .string "STUB_FUN" .text .globl STUB_FUN .type STUB_FUN,@functionSTUB_FUN:.LFB15: endbr64 pushq %rbp movq %rsp,%rbp sub......
  • AcWing 799:最长连续不重复子序列 ← 双指针
    【题目来源】https://www.acwing.com/problem/content/801/【题目描述】给定一个长度为n的整数序列,请找出最长的不包含重复的数的连续区间,输出它的长度。【输入格式】第一行包含整数n。第二行包含n个整数(均在0∼10^5范围内),表示整数序列。【输出格式】共一行,包含一个......
  • python毕设 基于银川九中的校园一卡通系统的设计与实现zopo58z6程序+论文 可用于毕业
    本系统(程序+源码+数据库+调试部署+开发环境)带论文文档1万字以上,文末可获取,系统界面在最后面。系统程序文件列表开题报告内容研究背景随着信息技术的不断进步和智能化校园建设的推进,校园一卡通系统已成为现代学校管理的重要组成部分。银川九中作为一所具有深厚文化底蕴和较......
  • python毕设 家居管理平台068mz1d9程序+论文 可用于毕业设计
    本系统(程序+源码+数据库+调试部署+开发环境)带论文文档1万字以上,文末可获取,系统界面在最后面。系统程序文件列表开题报告内容研究背景随着物联网技术的快速发展和智能家居产品的普及,家居管理已经成为现代家庭生活中的重要一环。然而,目前市场上的智能家居产品种类繁多,品牌各......
  • python毕设 健身房会员服务管理系统r226ooxg程序+论文 可用于毕业设计
    本系统(程序+源码+数据库+调试部署+开发环境)带论文文档1万字以上,文末可获取,系统界面在最后面。系统程序文件列表开题报告内容研究背景随着健身行业的蓬勃发展,健身房已成为现代人追求健康生活的重要场所。然而,传统的健身房管理方式往往依赖于人工记录和服务,这不仅效率低下,而......
  • python毕设 健身工作室管理系统w1k95gr1程序+论文 可用于毕业设计
    本系统(程序+源码+数据库+调试部署+开发环境)带论文文档1万字以上,文末可获取,系统界面在最后面。系统程序文件列表开题报告内容研究背景随着健康意识的日益增强,健身工作室作为一种提供个性化、高质量健身服务的场所,受到了越来越多人的青睐。然而,传统的健身工作室管理方式往往......
  • 力扣1. 两数之和
    相信很多人和我一样(本人较......
  • 后台管理系统动态面包屑Breadcrumb组件的实现
    在后管理系统开发中,面包屑导航是一个非常常见的功能,通常是根据当前的url自动生成面包屑导航菜单,当跳转路由发生变化时,面包屑导航都会随之发生变化,即动态面包屑。要完成动态面包屑我们需要制作一个动态数组,数组中每个item都表示一个路由信息,在这里我们使用到route.match......
  • 我也要学高代
    现在是1月5日,八点零二分。高等线性代数,一章没看。明天早九考试,何如?行列式\(|A|\)为行列式,\(M_{i,j}\)为余子式,\(A_{i,j}\)为代数余子式,有\((-1)^{i+j}\)可按照任意行或列展开:\(|A|=\sum\limits_{i=1}^{n}a_{r,i}A_{r,i}\)行列式的组合定义:\(|A|=\sum_P(-1)^{\tau(P......
  • 使用Pandoc复制ChatGPT生成的高数公式文本
    在现代的学习和研究中,数学公式的表达和共享变得尤为重要。随着ChatGPT等智能工具的崛起,我们能够轻松生成复杂的数学公式。然而,如何将这些公式导出、共享或进一步处理成为了一个新的挑战。Pandoc作为一个功能强大的文档转换工具,提供了一个解决方案,使得我们能够更方便地处理和......