提供了编程的基础技术教程

zl程序教程

您现在的位置是：首页 >  后端

当前栏目

Python进阶42-drf框架(四)

Python框架 进阶 42 drf
2023-06-13 09:11:06 时间

-曾老湿, 江湖人称曾老大。

-多年互联网运维工作经验，曾负责过大规模集群架构自动化运维管理工作。 -擅长Web集群架构与自动化运维，曾负责国内某大型金融公司运维工作。 -devops项目经理兼DBA。 -开发过一套自动化运维平台（功能如下）： 1)整合了各个公有云API，自主创建云主机。 2)ELK自动化收集日志功能。 3)Saltstack自动化运维统一配置管理工具。 4)Git、Jenkins自动化代码上线及自动化测试平台。 5)堡垒机，连接Linux、Windows平台及日志审计。 6)SQL执行及审批流程。 7)慢查询日志分析web界面。

三大认证准备工作

创建项目



创建数据库

mysql> create database dg_proj2;

设置数据库

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'dg_proj2',
        'HOST':'10.0.0.51',
        'USER':'zls',
        'PASSWORD':'123'
        # 'NAME': os.path.join(BASE_DIR, 'db.sqlite3'),
    }
}

使用pymysql

import pymysql
pymysql.install_as_MySQLdb()

设置时区

LANGUAGE_CODE = 'zh-hans'

TIME_ZONE = 'Asia/Shanghai'

USE_I18N = True

USE_L10N = True

USE_TZ = False

路由分发

from django.conf.urls import url, include
from django.contrib import admin

from django.views.static import serve
from django.conf import settings
urlpatterns = [
    url(r'^admin/', admin.site.urls),
    # 路由分发
    url(r'^api/', include('api.urls')),
]

设置Media及路由

MEDIA_URL = '/media/'
MEDIA_ROOT = os.path.join(BASE_DIR,'media')
from django.conf.urls import url, include
from django.contrib import admin

from django.views.static import serve
from django.conf import settings
urlpatterns = [
    url(r'^admin/', admin.site.urls),
    # 路由分发
    url(r'^api/', include('api.urls')),

    url(r'^media/(?P<path>.*)', serve, {'document_root': settings.MEDIA_ROOT}),
]

自定义exception

REST_FRAMEWORK = {
    'EXCEPTION_HANDLER': 'utils.exception.exception_handler',
}

自定义Response

from rest_framework.response import Response

"""
Response({
    'status':0,
    'msg':'ok',
    'results':[],
    'token':'', # 有 这样的额外的key-value数据结果
},status=http_status,headers=headers,exceptio=True|False)


## 我们想要的结果，就是在APIResponse上加上括号就出来最简单的结果
APIResponse() => Respone({'status':0, 'msg':'ok',})
"""


class APIResponse(Response):
    def __init__(self, data_status=0, data_msg='ok', results=None, http_status=None, headers=None, exception=False,
                 **kwargs):
        data = {
            'status': data_status,
            'msg': data_msg
        }

        ## results可能是False、0等数据，这些数据某些情况下，也会作为合法数据返回
        if results is not None:
            data['results'] = results

        if kwargs is not None:
            for k, v in kwargs.items():
                setattr(data, k, v)  ## 反射，相当于：data[k] = v
        super().__init__(data=data, status=http_status, headers=headers, exception=exception)

注册drf

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',
]

三大认证介绍

"""
1）APIView的dispath(self, request, *args, **kwargs)
2）dispath方法内 self.initial(request, *args, **kwargs) 进入三大认证
    # 认证组件：校验用户 - 游客、合法用户、非法用户
    # 游客：代表校验通过，直接进入下一步校验（权限校验）
    # 合法用户：代表校验通过，将用户存储在request.user中，再进入下一步校验（权限校验）
    # 非法用户：代表校验失败，抛出异常，返回403权限异常结果
    self.perform_authentication(request)
    
    # 权限组件：校验用户权限 - 必须登录、所有用户、登录读写游客只读、自定义用户角色
    # 认证通过：可以进入下一步校验（频率认证）
    # 认证失败：抛出异常，返回403权限异常结果
    self.check_permissions(request)
    
    # 频率组件：限制视图接口被访问的频率次数 - 限制的条件(IP、id、唯一键)、频率周期时间(s、m、h)、频率的次数（3/s）
    # 没有达到限次：正常访问接口
    # 达到限次：限制时间内不能访问，限制时间达到后，可以重新访问
    self.check_throttles(request)
    
"""

权限六表分析

基于用户权限访问的认证： RBAC （Role Based Acess Control）

自己简单了解：基于auth的认证规则

Django框架采用的是RBAC认证规则，RBAC认证规则通常分为三表规则、五表规则，Django采用的是六表规则

三表规则： 1.用户表 2.角色表 3.权限表

五表规则： 1.用户表 2.角色表 3.权限表 4.用户角色关联表 5.角色权限关联表

Django六表规则： 1.用户表 2.角色表 3.权限表 4.用户角色关联表 5.角色权限关联表 6.用户权限关联表

自定义权限六表

settings.py

## 配置自定义user表
AUTH_USER_MODEL = 'api.User'

models.py

from django.db import models

## 用户表：角色groups     权限user_permissions
## 角色表：用户user_set   权限permissions
## 权限表：用户user_set   角色group_set

from django.contrib.auth.models import AbstractUser

class User(AbstractUser):
    mobile = models.CharField(max_length=11,unique=True)

    class Meta:
        db_table = 'api_user'
        verbose_name = '用户表'
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.username

数据库迁移 

MacBook-pro:dg_proj_n driverzeng$ python3 manage.py makemigrations
MacBook-pro:dg_proj_n driverzeng$ python3 manage.py migrate



六表数据测试

创建超级用户

MacBook-pro:dg_proj_n driverzeng$ python3 manage.py createsuperuser
Username: admin
Email address:
Password:   admin12345
Password (again): admin12345
Superuser created successfully.

Django脚本文件

import os,django

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "dg_proj_n.settings")
django.setup()

六表数据关联测试

import os,django

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "dg_proj_n.settings")
django.setup()

from api import models
user = models.User.objects.first()
print(user.username)
print(user.groups.first().name)
print(user.user_permissions.first().name)

from django.contrib.auth.models import Group

group = Group.objects.first()

print(group.name)
print(group.user_set.first().username)
print(group.permissions.first().name)

from django.contrib.auth.models import Permission
p_16 = Permission.objects.filter(pk=15).first()
print(p_16.user_set.first().username)
p_17 = Permission.objects.filter(pk=17).first()
print(p_17.group_set.first().name)



认证组件

源码分析

"""
#################### 认证组件 ####################
    Request类的 方法属性 user 的get方法 => self._authenticate() 完成认证
    
    认证的细则：
    # 做认证
    def _authenticate(self):
        # 遍历拿到一个个认证器，进行认证
        # self.authenticators配置的一堆认证类产生的认证类对象组成的 list
        for authenticator in self.authenticators:
            try:
                # 认证器(对象)调用认证方法authenticate(认证类对象self, request请求对象)
                # 返回值：登陆的用户与认证的信息组成的 tuple
                # 该方法被try包裹，代表该方法会抛异常，抛异常就代表认证失败
                user_auth_tuple = authenticator.authenticate(self)
            except exceptions.APIException:
                self._not_authenticated()
                raise

            # 返回值的处理
            if user_auth_tuple is not None:
                self._authenticator = authenticator
                # 如何有返回值，就将 登陆用户 与 登陆认证 分别保存到 request.user、request.auth
                self.user, self.auth = user_auth_tuple
                return
        # 如果返回值user_auth_tuple为空，代表认证通过，但是没有 登陆用户 与 登陆认证信息，代表游客
        self._not_authenticated()

#################### 权限组件 ########################
    self.check_permissions(request)
    认证细则：
    def check_permissions(self, request):
        # 遍历权限对象列表得到一个个权限对象(权限器)，进行权限认证
        for permission in self.get_permissions():
            # 权限类一定有一个has_permission权限方法，用来做权限认证的
            # 参数：权限对象self、请求对象request、视图类对象
            # 返回值：有权限返回True，无权限返回False
            if not permission.has_permission(request, self):
                self.permission_denied(
                    request, message=getattr(permission, 'message', None)
                )
"""

自定义认证类

在api目录下创建authentications.py文件

## 自定义认证类

# 1.继承BaseAuthentication
# 2.重写authenticate(self,request)方法，自定义认证规则
# 3.认证规则基于：
#   - 没有认证信息返回None（游客）
#   - 有认证信息认证失败抛异常（非法用户）
#   - 有认证信息认证成功返回用户与认证信息元组（合法用户）
# 4.完成视图类的配置：
#   - 全局（settings文件中）
#   - 局部（确切的视图类）
from rest_framework.exceptions import AuthenticationFailed
from rest_framework.authentication import BaseAuthentication
from . import models


class MyAuthentication(BaseAuthentication):
    def authenticate(self, request):
        # 前台在请求头中携带认证信息，且默认规范用Authorization字段携带认证信息
        # 后台固定在请求对象的META字典中获取
        auth = request.META.get('HTTP_AUTHORIZATION', None)

        # 处理游客
        if auth is None:
            return None
        ##  设置认证字段小规则（两段式）："auth 认证字符串"
        auth_list = auth.split()

        ## 校验合法还是非法用户
        if not (len(auth_list) == 2 and auth_list[0].lower() == 'auth'):
            raise AuthenticationFailed('认证信息有误：非法用户')
        ## 合法的用户还需要从auth_list[1]中解析出来
        ## 假设一种情况：信息为abc.123.xyz,就可以解析出admin用户；实际开发，该逻辑一定是校验用户的正常逻辑
        if auth_list[1] != 'abc.123.xyz':  # 校验失败
            raise AuthenticationFailed('用户校验失败：非法用户')

        user = models.User.objects.filter(username='admin').first()

        if not user:
            raise AuthenticationFailed('用户数据有误：非法用户')

        return (user, None)

全局配置

REST_FRAMEWORK = {
    'EXCEPTION_HANDLER': 'utils.exception.exception_handler',
    
    ### 配置自己写的认证类
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'api.authentications.MyAuthentication',
    ]
}

视图测试

from rest_framework.views import APIView
from rest_framework.generics import GenericAPIView
from rest_framework.viewsets import GenericViewSet, ViewSet
from utils.response import APIResponse


class TestView(APIView):
    def get(self, request, *args, **kwargs):
        return APIResponse(0, 'Test Get OK')

路由设置

from django.conf.urls import url
from . import views

urlpatterns = [
    url(r'^test/$', views.TestView.as_view())
]

没有带认证信息







如果认证通过，后台就可以从request.user中拿到用户

from rest_framework.views import APIView
from rest_framework.generics import GenericAPIView
from rest_framework.viewsets import GenericViewSet, ViewSet
from utils.response import APIResponse


class TestView(APIView):
    def get(self, request, *args, **kwargs):
        print(request.user)
        return APIResponse(0, 'Test Get OK')



认证组件的小结

"""
1) 创建继承BaseAuthentication的认证类
2) 实现authenticate方法
3) 实现体根据认证规则 确定游客、非法用户、合法用户
4) 进行全局或局部配置

认证规则
i.没有认证信息返回None(游客)
ii.有认证信息认证失败抛异常(非法用户)
iii.有认证信息认证成功返回用户与认证信息元组(合法用户)
"""

权限组件

"""
1）AllowAny：
    认证规则全部返还True：return True
        游客与登陆用户都有所有权限

2) IsAuthenticated：
    认证规则必须有登陆的合法用户：return bool(request.user and request.user.is_authenticated)
        游客没有任何权限，登陆用户才有权限
    
3) IsAdminUser：
    认证规则必须是后台管理用户：return bool(request.user and request.user.is_staff)
        游客没有任何权限，登陆用户才有权限

4) IsAuthenticatedOrReadOnly
    认证规则必须是只读请求或是合法用户：
        return bool(
            request.method in SAFE_METHODS or
            request.user and
            request.user.is_authenticated
        )
        游客只读，合法用户无限制
"""

登录后才能通过认证：视图

from rest_framework.views import APIView
from rest_framework.generics import GenericAPIView
from rest_framework.viewsets import GenericViewSet, ViewSet
from utils.response import APIResponse


class TestView(APIView):
    def get(self, request, *args, **kwargs):
        print(request.user)
        return APIResponse(0, 'Test Get OK')

## 只有登录后才能访问
from rest_framework.permissions import IsAuthenticated
class TestAuthenticatedAPIView(APIView):
    ## 局部认证
    permission_classes = [IsAuthenticated]
    def get(self, request, *args, **kwargs):
        # print(request.user)
        return APIResponse(0, '登陆成功 访问接口 OK')

登录后才能通过认证：路由

from django.conf.urls import url
from . import views

urlpatterns = [
    url(r'^test/$', views.TestView.as_view()),
    url(r'^test1/$', views.TestAuthenticatedAPIView.as_view())
]





游客只读：视图

from rest_framework.views import APIView
from rest_framework.generics import GenericAPIView
from rest_framework.viewsets import GenericViewSet, ViewSet
from utils.response import APIResponse


class TestView(APIView):
    def get(self, request, *args, **kwargs):
        print(request.user)
        return APIResponse(0, 'Test Get OK')

## 只有登录后才能访问
from rest_framework.permissions import IsAuthenticated
class TestAuthenticatedAPIView(APIView):
    ## 局部认证
    permission_classes = [IsAuthenticated]
    def get(self, request, *args, **kwargs):
        # print(request.user)
        return APIResponse(0, '登陆成功 访问接口 OK')

## 游客只读，增删改查必须登录
from rest_framework.permissions import IsAuthenticatedOrReadOnly
class TestAuthenticatedOrReadOnlyAPIView(APIView):
    ## 局部认证
    permission_classes = [IsAuthenticatedOrReadOnly]
    def get(self, request, *args, **kwargs):
        return APIResponse(0, '未登录，游客访问，读取 OK')

    def post(self,request,*args,**kwargs):
        return APIResponse(0, '已登录，登录用户，读写 OK')

游客只读：路由

from django.conf.urls import url
from . import views

urlpatterns = [
    url(r'^test/$', views.TestView.as_view()),
    url(r'^test1/$', views.TestAuthenticatedAPIView.as_view()),
    url(r'^test2/$', views.TestAuthenticatedOrReadOnlyAPIView.as_view()),
]





自定义权限类

需求：游客只读，登录用户只读，管理员可以增删改查。

权限类文件

在api目录下创建，permissions.py文件，在里面定义权限类

from rest_framework.permissions import BasePermission
from django.contrib.auth.models import Group


class MyPermission(BasePermission):
    def has_permission(self, request, view):
        r1 = request.method in ('GET', 'HEAD', 'OPTIONS')
        group = Group.objects.filter(name='管理员').first()
        groups = request.user.groups.all()
        r2 = group and groups
        r3 = group in groups
        return r1 or (r2 and r3)

视图

from rest_framework.views import APIView
from rest_framework.generics import GenericAPIView
from rest_framework.viewsets import GenericViewSet, ViewSet
from utils.response import APIResponse
from .permissions import MyPermission


class TestView(APIView):
    def get(self, request, *args, **kwargs):
        print(request.user)
        return APIResponse(0, 'Test Get OK')

## 需求：游客只读，登录用户只读，管理员可以增删改查。

from rest_framework.permissions import IsAuthenticatedOrReadOnly
class TestAdminOrReadOnlyAPIView(APIView):
    ## 局部认证
    permission_classes = [MyPermission]
    def get(self, request, *args, **kwargs):
        return APIResponse(0, '读取 OK')

    def post(self,request,*args,**kwargs):
        return APIResponse(0, '读写 OK')

路由

from django.conf.urls import url
from . import views

urlpatterns = [
    url(r'^test/$', views.TestView.as_view()),
    url(r'^test1/$', views.TestAuthenticatedAPIView.as_view()),
    url(r'^test2/$', views.TestAuthenticatedOrReadOnlyAPIView.as_view()),
    url(r'^test3/$', views.TestAdminOrReadOnlyAPIView.as_view()),
]









自定义权限类小结

"""
1) 创建继承BasePermission的权限类
2) 实现has_permission方法
3) 实现体根据权限规则 确定有无权限
4) 进行全局或局部配置

认证规则
i.满足设置的用户条件，代表有权限，返回True
ii.不满足设置的用户条件，代表有权限，返回False
"""

频率组件

频率类源码

入口

# 1）APIView的dispath方法中的 self.initial(request, *args, **kwargs) 点进去
# 2）self.check_throttles(request) 进行频率认证

# 频率组件核心源码分析
def check_throttles(self, request):
    throttle_durations = []
    # 1）遍历配置的频率认证类，初始化得到一个个频率认证类对象（会调用频率认证类的 __init__() 方法）
    # 2）频率认证类对象调用 allow_request 方法，判断是否限次（没有限次可访问，限次不可访问）
    # 3）频率认证类对象在限次后，调用 wait 方法，获取还需等待多长时间可以进行下一次访问
    # 注：频率认证类都是继承 SimpleRateThrottle 类
    for throttle in self.get_throttles():
        if not throttle.allow_request(request, self):
            # 只要频率限制了，allow_request 返回False了，才会调用wait
            throttle_durations.append(throttle.wait())

            if throttle_durations:
                # Filter out `None` values which may happen in case of config / rate
                # changes, see #1438
                durations = [
                    duration for duration in throttle_durations
                    if duration is not None
                ]

                duration = max(durations, default=None)
                self.throttled(request, duration)

自定义频率类

需求：短信接口 1/min 频率限制

步骤

# 1) 自定义一个继承 SimpleRateThrottle 类的频率类
# 2) 设置一个 scope 类属性，属性值为任意见名知意的字符串
# 3) 在settings配置文件中，配置drf的DEFAULT_THROTTLE_RATES，格式为 {scope字符串: '次数/时间'}
# 4) 在自定义频率类中重写 get_cache_key 方法
    # 限制的对象返回 与限制信息有关的字符串
    # 不限制的对象返回 None (只能放回None，不能是False或是''等)

throttles 频率类

首先先在api目录下创建一个throttles.py

from rest_framework.throttling import SimpleRateThrottle


class SMSRateThrottle(SimpleRateThrottle):
    scope = 'sms'

    ## 只对提交手机号的get方法进行限制
    def get_cache_key(self, request, view):
        mobile = request.query_params.get('mobile')
        ## 没有手机号 ，就不做 频率限制
        if not mobile:
            return None
        ## 返回可以根据手机号动态变化，且不易重复的字符串，作为操作缓存的key
        return 'throttle_%(scope)s_%(ident)s' % {'scope': self.scope, 'ident': mobile}

全局配置

settings.py配置

REST_FRAMEWORK = {
    # 全局配置异常模块
    'EXCEPTION_HANDLER': 'utils.exception.exception_handler',
    # 认证类配置
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.SessionAuthentication',
        'rest_framework.authentication.BasicAuthentication',
        'api.authentications.MyAuthentication',
    ],
    # 权限类配置
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.AllowAny',
    ],
    # 频率类配置
    'DEFAULT_THROTTLE_CLASSES': [],
    # 频率限制条件配置
    'DEFAULT_THROTTLE_RATES': {
        'user': '3/min',
        'anon': None,
        'sms': '3/min'
    },
}

局部配置

创建view.py视图

一般我们都会走局部配置

from rest_framework.views import APIView
from rest_framework.generics import GenericAPIView
from rest_framework.viewsets import GenericViewSet, ViewSet
from utils.response import APIResponse
from .permissions import MyPermission
from . throttles import SMSRateThrottle

class TestSMSAPIView(APIView):
    throttle_classes = [SMSRateThrottle]
    def get(self, request, *args, **kwargs):
        return APIResponse(0, 'get 获取验证码 OK')

    def post(self,request,*args,**kwargs):
        return APIResponse(0, 'post 获取验证码 OK')

路由

from django.conf.urls import url
from . import views

urlpatterns = [
    url(r'^test/$', views.TestView.as_view()),
    url(r'^test1/$', views.TestAuthenticatedAPIView.as_view()),
    url(r'^test2/$', views.TestAuthenticatedOrReadOnlyAPIView.as_view()),
    url(r'^test3/$', views.TestAdminOrReadOnlyAPIView.as_view()),
    url(r'^sms/$', views.TestSMSAPIView.as_view()),
]





相关文章