Django Rest Framework(二)

•基于Django

先创建一个django项目，在项目中创建一些表，用来测试rest framework的各种组件

models.py

class UserInfo(models.Model):
    """用户信息表"""
    user = models.CharField(max_length=32)
    pwd = models.CharField(max_length=32)


class UserToken(models.Model):
    """用户token表"""
    user = models.OneToOneField(to="UserInfo", on_delete=models.CASCADE)
    token = models.CharField(max_length=64)


class Courses(models.Model):
    """
    课程表
    """
    name = models.CharField(verbose_name="课程名称", max_length=32)
    course_img = models.CharField(verbose_name="课程图片", max_length=64)
    level_choices = (
        (1, "初级"),
        (2, "中级"),
        (3, "高级"),
    )
    level = models.IntegerField(verbose_name="难度", choices=level_choices, default=1)

    def __str__(self):
        return self.name


class CourseDetail(models.Model):
     """课程详细表"""
     course = models.OneToOneField(to="Courses", on_delete=models.CASCADE)
     slogan = models.CharField(verbose_name="口号", max_length=255)
     why = models.CharField(verbose_name="为什么要学", max_length=255)
     recommend_courses = models.ManyToManyField(verbose_name="推荐课程", to="Courses",
                                                related_name="rc")  # related_name设置反向查询的字段，有多个关联时指定某个字段进行反向查询

    def __str__(self):
         return "课程详细：" + self.course.title



class Chapter(models.Model):
    """
    课程章节表
    """
    num = models.IntegerField(verbose_name="章节")
    name = models.CharField(verbose_name="章节名称", max_length=32)
    course = models.ForeignKey(verbose_name="所属课程", to="Courses", related_name='coursechapters', on_delete=models.CASCADE)

    def __str__(self):
        return self.name

  

urls.py

from django.contrib import admin
from django.urls import path, include, re_path

urlpatterns = [
    path('admin/', admin.site.urls),
    re_path(r'^api/(?P<version>w+)/', include("api.urls")),
]

api/urls.py

from django.urls import re_path
from api.views import courses, account, micro

urlpatterns = [
    re_path(r'^courses/$', courses.CoursesView.as_view({"get": "list"})),
    re_path(r'^courses/(?P<pk>d+)/', courses.CoursesView.as_view({"get": "retrieve"})),

    re_path(r'^micro/$', micro.MicroView.as_view({"get": "list"})),

    re_path(r'^login/$', account.LoginView.as_view()),
]

•基本流程

请求到来之后，都要执行dispatch方法，dispatch方法根据请求方式不同触发不同的方法，返回不同的内容

url.py

from django.conf.urls import url, include
from api.views import TestView
 
urlpatterns = [
    url(r'^test/', TestView.as_view()),
]

views.py

from rest_framework.views import APIView
from rest_framework.response import Response
 
 
class TestView(APIView):
    def dispatch(self, request, *args, **kwargs):
        """
        请求到来之后，都要执行dispatch方法，dispatch方法根据请求方式不同触发 get/post/put等方法
         
        注意：dispatch方法有好多好多的功能
        """
        return super().dispatch(request, *args, **kwargs)
 
    def get(self, request, *args, **kwargs):
        return Response('GET请求，响应内容')
 
    def post(self, request, *args, **kwargs):
        return Response('POST请求，响应内容')
 
    def put(self, request, *args, **kwargs):
        return Response('PUT请求，响应内容')

•认证组件

给micro添加需要认证才能访问的权限

micro.py

# _*_ coding=utf-8 _*_
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.viewsets import ViewSetMixin
from api.auth.auth import MicroAuth


class MicroView(ViewSetMixin, APIView):
    # 给micro添加认证后才能访问的组件
    authentication_classes = [MicroAuth]

    def list(self, request, *args, **kwargs):
        ret = {'code': 1000, 'data': '学习中心'}

        return Response(ret)

auth.py

# _*_ coding=utf-8 _*_
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from api.models import *


class MicroAuth(BaseAuthentication):
    """从token表读取token进行认证"""
    def authenticate(self, request):
        token = request.query_params.get('token')
        obj = UserToken.objects.filter(token=token).first()
        if not obj:
            raise AuthenticationFailed({'code': 10001, 'error': '认证失败'})
        return (obj.user.user, obj)

全局设置：

上述操作中均是对单独视图进行特殊配置，如果想要对全局进行配置，则需要再配置文件中写入即可

•访问频率控制

给course添加频率限制，同一个IP，60秒内访问不超过3次

•使用自定义类进行访问频率控制，继承BaseThrottle类

myThrottle.py

import time
from rest_framework.throttling import BaseThrottle


class MyThrottle(BaseThrottle):
    """IP访问频率组件
    限制60秒内访问3次"""

    def __init__(self):
        self.history = None

    def allow_request(self, request, view):
        current_time = time.time()

        ip = request.META.get('REMOTE_ADDR')
        print(ip)

        if ip not in visit_code:
            # 如果是第一次访问就把此ip的访问时间存入visit_code中，返回True，不限制
            visit_code[ip] = [current_time, ]
            return True

        # 如果不是第一次访问，就获取其ip的访问时间[time1,time2..]
        history = visit_code.get(ip)
        self.history = history
        # print(history,visit_code)

        while history and history[-1] < current_time - 60:
            # 判断第一次访问时间和当前时间是否超过60s，超过则删除
            history.pop()

        if len(history) < 3:
            # history里面的元素小于3个则把当前时间添加进去，方法True
            history.insert(0, current_time)
            return True
        # else:  # 可以不写
        #     return False

    def wait(self):
        """需要等待多少时间才能访问"""
        current_time = time.time()
        return 60 - (current_time - self.history[-1])

在CoursesView类添加访问频率组件

class CoursesView(ViewSetMixin, APIView):
    # 频率访问组件
    throttle_classes = [MyThrottle,]

•使用rest framework内置频率控制组件

myThrottle.py

from rest_framework.throttling import  SimpleRateThrottle


class VisitThrottle(SimpleRateThrottle):
    """内置ip频率组件,需要在settings里面设置参数"""
    scope = "visit_rate"

    def get_cache_key(self, request, view):
        return self.get_ident(request)

settings.py

REST_FRAMEWORK = {
    "DEFAULT_THROTTLE_CLASSES": ["api.myThrottle.VisitThrottle", ],
    "DEFAULT_THROTTLE_RATES": {
        "visit_rate": "5/m",
        # 这个参数就是频率类中定义的那个参数scope， 其中第一个数字5表示5次，
        # 后面的m表示一分钟，还有s，一秒， h, 一小时， d, 一天
    }
}

•权限控制

给TestView添加权限认证

views.py

# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import BasePermission
from rest_framework.request import Request


class TestPermission(BasePermission):
    message = "权限验证失败"

    def has_permission(self, request, view):
        """
        判断是否有权限访问当前请求
        Return `True` if permission is granted, `False` otherwise.
        :param request: 
        :param view: 
        :return: True有权限；False无权限
        """
        if request.user == "管理员":
            return True

    # GenericAPIView中get_object时调用
    def has_object_permission(self, request, view, obj):
        """
        继承GenericAPIView，并在其中使用get_object时获取对象时，触发单独对象权限验证
        Return `True` if permission is granted, `False` otherwise.
        :param request: 
        :param view: 
        :param obj: 
        :return: True有权限；False无权限
        """
        if request.user == "管理员":
            return True


class TestView(APIView):
    # 认证的动作是由request.user触发
    # 添加权限
    permission_classes = [TestPermission, ]

    def get(self, request, *args, **kwargs):
        # self.dispatch
        print(request.user)
        print(request.auth)
        return Response('GET请求，响应内容')

全局设置

settings.py

REST_FRAMEWORK = {
        "DEFAULT_PERMISSION_CLASSES": [
        "api.views.TestPermission",
    ],
}

•序列化

对用户请求的数据进行序列化

a.自动生成字段

# _*_ coding=utf-8 _*_
from django.urls import re_path
from api.views import courses


urlpatterns = [
    re_path(r'^courses/$', courses.CoursesView.as_view({"get": "list"})),
]

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.viewsets import ViewSetMixin
from rest_framework import serializers
from api.models import *


class CoursesSerializers(serializers.ModelSerializer):
    """课程序列化"""
    level = serializers.CharField(source='get_level_display')   # 获取choices的中文

        class Meta:
        model = Course  # 指定表
        fields = "__all__"  # 使用自动生成的字段


class CoursesView(ViewSetMixin, APIView):

    def list(self, request, *args, **kwargs):
        ret = {"code": 1000, "data": None}
        try:
            course_list = Course.objects.all()
            course_data = CoursesSerializers(instance=course_list, many=True)
            ret["data"] = course_data.data
        except Exception as e:
            ret["code"] = 1001
            ret["error"] = "获取失败"

        return Response(ret)

b.使用自定义字段

# _*_ coding=utf-8 _*_
from django.urls import re_path
from api.views import courses

urlpatterns = [
    re_path(r'^courses/$', courses.CoursesView.as_view({"get": "list"})),
    re_path(r'^courses/(?P<pk>d+)/', courses.CoursesView.as_view({"get": "retrieve"})),
]

from rest_framework import serializers
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.viewsets import ViewSetMixin
from api.serializers.courses_serializers import *
from api.models import *


class CourseDetailSerializers(serializers.ModelSerializer):
    """课程详细序列化"""
    # OneToOne/fk/choice等字段查询某一条记录时适用： 自定义要序列化的字段，关联的表：表名.字段,
    name = serializers.CharField(source='course.name')
    level = serializers.CharField(source='course.get_level_display')  # 获取choice类型对应的中文

    # ManyToMany，fk查询多条记录适用:
    recommend_courses = serializers.SerializerMethodField()
    chapters = serializers.SerializerMethodField()

    class Meta:
        model = CourseDetail
        # 添加自定义字段
        fields = ['name', 'level',  'recommend_courses', 'chapters']

    def get_recommend_courses(self, obj):
        # 获取所有的课程,obj指的是CourseDetail,recommends字段数据为此函数的返回值
        course_list = obj.recommend_course.all()
        return [{"id": i.id, "title": i.name} for i in course_list]

    def get_chapters(self, obj):
        # 反向查询时如果定义了related_name，要使用related_name的值进行查询
        chapters = obj.course.coursechapters.all()
        return [{"id": i.id, "name": i.name} for i in chapters]


class CoursesView(ViewSetMixin, APIView):

    def list(self, request, *args, **kwargs):
        ret = {"code": 1000, "data": None}
        try:
            course_list = Course.objects.all()
            course_data = CoursesSerializers(instance=course_list, many=True)
            ret["data"] = course_data.data
        except Exception as e:
            ret["code"] = 1001
            ret["error"] = "获取失败"

        return Response(ret)

    def retrieve(self, request, *args, **kwargs):
        ret = {"code": 1000, "data": None}
        try:
            # 课程id
            pk = kwargs.get("pk")
            # 课程详细对象
            obj = CourseDetail.objects.filter(course_id=pk).first()
            course_detail = CourseDetailSerializers(instance=obj, many=False)
            ret["data"] = course_detail.data
        except Exception as e:
            ret["code"] = 1001
            ret["error"] = "获取失败"
        return Response(ret)

•分页器

设置页码进行分页

# _*_ coding=utf-8 _*_
from django.urls import re_path
from api.views import courses

urlpatterns = [
    re_path(r'^courses/$', courses.CoursesView.as_view({"get": "list"})),
    re_path(r'^courses/(?P<pk>d+)/', courses.CoursesView.as_view({"get": "retrieve"})),
]

# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework import serializers
from api import models

from rest_framework.pagination import PageNumberPagination


class MyPagination(PageNumberPagination):
    # 默认每页显示的数据条数
    page_size = 1
    # 获取URL参数中设置的每页显示数据条数
    page_size_query_param = 'page_size'

    # 获取URL参数中传入的页码key
    page_query_param = 'page'

    # 最大支持的每页显示的数据条数
    max_page_size = 1


class CourseSerializer(serializers.ModelSerializer):
    class Meta:
        model = models.Course
        fields = "__all__"


class CoursesViewSet(APIView):
    def list(self, request, *args, **kwargs):
        course_list = models.Course.objects.all().order_by('-id')

        # 实例化分页对象，获取数据库中的分页数据
        paginator = MyPagination()
        page_course_list = paginator.paginate_queryset(course_list, self.request, view=self)

        # 序列化对象
        serializer = CourseSerializer(page_user_list, many=True)

        # 生成分页和数据
        response = paginator.get_paginated_response(serializer.data)
        return response

使用自动生成url

urls.py

from rest_framework import routers
from api.views import *

router = routers.DefaultRouter()
router.register('courses', CoursesViewSet)

便捷使用分页器

views.py

from rest_framework import viewsets
from rest_framework import serializers
from api.models import *


class CoursesSerializers(serializers.ModelSerializer):
    """课程序列化"""
    level = serializers.CharField(source='get_level_display')   
        class Meta:
        model = Courses  
        fields = "__all__"  


class CoursesViewSet(viewsets.ModelViewSet):
     # 指定已经写好的分页器
    pagination_class = MyPagination
   
    queryset = Courses.objects.all().order_by('id')  # 使用分页器要排序
    serializer_class = CoursesSerializersModel