django rest_framework中的APIView,ModelViewSet,認證，權限，頻率，版本

時間 2019-11-19

標籤 django rest framework apiview modelviewset 認證權限頻率版本欄目 Python 简体版

原文原文鏈接

models.py

from django.db import models

class UserInfo(models.Model):
    name=models.CharField(max_length=32)
    pwd=models.CharField(max_length=32)
    type_choices=((1,"普通用戶"),(2,"VIP"),(3,"SVIP"))
    user_type=models.IntegerField(choices=type_choices,default=1)


class Token(models.Model):
    user=models.OneToOneField("UserInfo")
    token = models.CharField(max_length=128)

    def __str__(self):
        return self.token


class Book(models.Model):
    title=models.CharField(max_length=32)
    price=models.IntegerField()
    pub_date=models.DateField()
    publish=models.ForeignKey("Publisher")
    authors=models.ManyToManyField("Author")
    def __str__(self):
        return self.title

class Publisher(models.Model):
    name=models.CharField(max_length=32)
    email=models.EmailField()
    def __str__(self):
        return self.name

class Author(models.Model):
    name=models.CharField(max_length=32)
    age=models.IntegerField()
    def __str__(self):
        return self.name

views.pyhtml

from django.shortcuts import render

from app01 import models
from rest_framework import serializers
from rest_framework.response import Response
from rest_framework.views import APIView

class PublisherModelSerializer(serializers.ModelSerializer):
    class Meta:
        model = models.Publisher
        fields = "__all__"

class BookModelSerializer(serializers.ModelSerializer):
    class Meta:
        model = models.Book
        fields = "__all__"

class AuthorModelSerializer(serializers.ModelSerializer):
    class Meta:
        model = models.Author
        fields = "__all__"

#****用戶登陸認證組件
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed

class AuthToken(BaseAuthentication):

    def authenticate(self, request):
        token = request.query_params.get('token')   #query_params函數返回的是self._request.GET
        obj = models.Token.objects.filter(token=token).first()
        if not obj:
            raise AuthenticationFailed({'code':1001,'error':'認證失敗'})
        return (obj.user.name,obj)

#*********權限組件****
from rest_framework.permissions import BasePermission
class SVIPPermission(BasePermission):  #繼承於BasePermission，/authous/2/才能查看每一個具體的做者。不然報錯
    message = "只有超級用戶才能訪問"
    def has_permission(self,request,view):
        username = request.user
        print(username)
        user_type = models.UserInfo.objects.filter(name=username).first().user_type
        if user_type == 3:
            return True
        else:
            return False
##*******頻率組件
# from rest_framework.throttling import BaseThrottle
# VISIT_RECORD={}
# class VisitThrottle(BaseThrottle):
#
#     def __init__(self):
#         self.history=None
#
#     def allow_request(self,request,view):
#         remote_addr = request.META.get('REMOTE_ADDR')
#         print(remote_addr)
#         import time
#         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)<10:
#             history.insert(0,ctime)
#             return True
#         else:
#             return False
#
#     def wait(self):
#         import time
#         ctime=time.time()
#         return 60-(ctime-self.history[-1])


class PublisherView(APIView):

    #獲取全部出版社
    def get(self,request):

        pub_list = models.Publisher.objects.all()
        pub_ser = PublisherModelSerializer(pub_list,many=True)
        return Response(pub_ser.data)
    #提交出版社
    def post(self,request):

        pub_ser = PublisherModelSerializer(data=request.data)
        if pub_ser.is_valid():
            print(pub_ser.validated_data)
            pub_ser.save()
            return Response(pub_ser.data)
        else:
            return Response(pub_ser.errors)

class PublisherDetailView(APIView):
    #查看某個出版社
    def get(self,request,pk):
        pub_obj = models.Publisher.objects.filter(pk=pk).first()
        pub_ser = PublisherModelSerializer(pub_obj)
        return Response(pub_ser.data)
    #更新某個出版社
    def put(self,request,pk):
        pub_obj = models.Publisher.objects.filter(pk=pk).first()
        pub_ser = PublisherModelSerializer(pub_obj,data=request.data)
        if pub_ser.is_valid():
            pub_ser.save()
            return Response(pub_ser.data)
        else:
            return Response(pub_ser.errors)
    #刪除某個出版社
    def delete(self,request,pk):
        models.Publisher.objects.filter(pk=pk).delete()
        return Response()

#****解析器
###--->根據contenttype的不一樣解析成不一樣找到不一樣的解析器，解析出的是字典或者QueryDict
##不管post(請求)前面傳的是JSON數據仍是urlencoded數據，rest framework 中的request.data都
# 幫咱們解析成字典  而django中的request.POST只解析urlencode數據
from rest_framework.parsers import JSONParser,FormParser,MultiPartParser,FileUploadParser
#默認使用前三個解析器
# 在視圖類中添加 parser_classes = [JSONPaser,] #若是隻寫這一個，意味着只能解析JSON數據  FormParser解析urlencode數據

#*********渲染器*****
from rest_framework.renderers import JSONRenderer,BrowsableAPIRenderer
#在視圖類中添加 renderer_classes = [JSONRenderer,]

#****響應器就是Response比django原生的HttpResponse多了一些功能

#*******版本***
from rest_framework.versioning import QueryParameterVersioning,URLPathVersioning
# versioning_class = URLPathVersioning  request.version,獲取版本的值推薦用URLPathVersioning不用QueryParameterVersioning

# # ******第二種分頁   *********
from rest_framework.pagination import PageNumberPagination

class MyPageNumberPagination(PageNumberPagination):

    page_size = 2  # 每頁顯示多少條數據
    page_query_param = "page"  # URL中輸入的 page=2
    page_size_query_param = "size"  # 臨時瀏覽器中使用127.0.0.1:8000/books/?page=2&size=2 第二頁顯示兩條數據
    max_page_size = 3  # 上面的size最大值，不能無限制顯示10000條數據

# # ******第三種分頁   *********
from rest_framework.pagination import PageNumberPagination,LimitOffsetPagination
class MyLimitOffsetPagination(LimitOffsetPagination):
    default_limit = 1 #每頁顯示多少條數據

class BookView(APIView):

    # authentication_classes = [AuthToken,]
    # permission_classes = [SVIPPermission,] #依賴於上面的AuthToken
    # throttle_classes = [] #頻率
    # parser_classes = [JSONParser, FormParser] #解析器
    # renderer_classes = [JSONRenderer,]  #渲染器JSONRenderer只返回json數據，不帶html頁面
    # versioning_class = QueryParameterVersioning

    # def get(self,request):
    #
    #     book_list = models.Book.objects.all()
    #
    #     from rest_framework.pagination import PageNumberPagination
    #     # ******第一種分頁   不經常使用*********
    #     #在settings中配置"PAGE_SIZE"的值,定義多少條數據就顯示多少 不經常使用
    #     pnp = PageNumberPagination()
    #     book_pag = pnp.paginate_queryset(book_list,request,self)
    #     book_ser = BookModelSerializer(book_pag,many=True)
    #     return Response(book_ser.data)

    def get(self,request,*args,**kwargs):

        book_list = models.Book.objects.all()
        #**第二種分頁
        pnp = MyPageNumberPagination()
        book_pag = pnp.paginate_queryset(book_list,request,self)
        book_ser = BookModelSerializer(book_pag,many=True)

        return Response(book_ser.data)

    # def get(self,request):
    #
    #     book_list = models.Book.objects.all()
    #     #**第三種分頁
    #     pnp = MyLimitOffsetPagination()
    #     book_pag = pnp.paginate_queryset(book_list,request,self)
    #     book_ser = BookModelSerializer(book_pag,many=True)
    #
    #     return Response(book_ser.data)

    def post(self,request,*args,**kwargs):

        book_ser = BookModelSerializer(data=request.data)
        if book_ser.is_valid():
            print(book_ser.validated_data)
            book_ser.save()
            return Response(book_ser.data)
        else:
            return Response(book_ser.errors)


class BookDetailView(APIView):

    def get(self,request,pk,*args,**kwargs):
        book_obj = models.Book.objects.filter(pk=pk).first()
        book_ser = BookModelSerializer(book_obj)
        return Response(book_ser.data)

    def put(self,request,pk,*args,**kwargs):
        book_obj = models.Book.objects.filter(pk=pk).first()
        book_ser = BookModelSerializer(book_obj,data=request.data)
        if book_ser.is_valid():
            book_ser.save()
            return Response(book_ser.data)
        else:
            return Response(book_ser.errors)

    def delete(self,request,pk,*args,**kwargs):
        models.Book.objects.filter(pk=pk).delete()
        return Response()
##***局部的頻率組件
from app01.utils import VisitThrottle

from rest_framework import viewsets
class BookViewSet(viewsets.ModelViewSet):
    # authentication_classes = [AuthToken, ]
    # permission_classes=[SVIPPermission,]
    throttle_classes = [VisitThrottle]    # 限制某個IP每分鐘訪問次數不能超過20次
    queryset = models.Book.objects.all()
    serializer_class = BookModelSerializer
    #ModelViewSet中的分頁
    pagination_class = MyPageNumberPagination


import uuid
class AuthView(APIView):
    #authentication_classes = []  # 全局配置後，不讓login驗證，直接加空列表就不驗證
    def post(self,request,*args,**kwargs):

        ##用戶登陸認證

        ret = {'code':1000}
        user = request.data.get('user')
        pwd = request.data.get('pwd')

        user = models.UserInfo.objects.filter(user=user,pwd=pwd).first()
        if not user:
            ret['code'] = 1001
            ret['error'] = '用戶名或密碼錯誤'
        else:
            uid = str(uuid.uuid4())
            models.Token.objects.update_or_create(user=user,defaults={'token':uid})
            ret['token'] = uid
        return Response(ret)

utils.py 認證，權限，頻率

from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from app01 import models

#*********用戶登陸認證組件****
class AuthToken(BaseAuthentication):

    def authenticate(self, request):
        token = request.query_params.get('token')
        obj = models.Token.objects.filter(token=token).first()
        if not obj:
            raise AuthenticationFailed({'code':1001,'error':'認證失敗'})
        return (obj.user.name,obj)

#*********權限組件****
from rest_framework.permissions import BasePermission
class SVIPPermission(BasePermission):  #繼承於BasePermission，/authous/2/才能查看每一個具體的做者。不然報錯
    message = "只有超級用戶才能訪問"
    def has_permission(self,request,view):
        username = request.user
        print(username)
        user_type = models.UserInfo.objects.filter(name=username).first().user_type
        if user_type == 3:
            return True
        else:
            return False


#*********頻率組件****
from rest_framework.throttling import BaseThrottle
VISIT_RECORD={}
class VisitThrottle(BaseThrottle):

    def __init__(self):
        self.history=None

    def allow_request(self,request,view):
        remote_addr = request.META.get('REMOTE_ADDR')
        print(remote_addr)
        import time
        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)<10:
            history.insert(0,ctime)
            return True
        else:
            return False

    def wait(self):
        import time
        ctime=time.time()
        return 60-(ctime-self.history[-1])

setting.pydjango

REST_FRAMEWORK = {

    # "DEFAULT_AUTHENTICATION_CLASSES": ["app01.utils.TokenAuth",],
    # "DEFAULT_PERMISSION_CLASSES": ["app01.utils.SVIPPermission",],
    # "DEFAULT_THROTTLE_CLASSES":["app01.utils.VisitThrottle",],
    #'DEFAULT_RENDERER_CLASSES':['rest_framework.renderers.JSONRenderer','rest_framework.renderers.BrowsableAPIRenderer',],
    # "PAGE_SIZE":2,
    'DEFAULT_VERSIONING_CLASS':'rest_framework.versioning.URLPathVersioning',
    'ALLOWED_VERSIONS':['v1','v2'], # 容許的版本
    'VERSION_PARAM':'version', # 參數
    'DEFAULT_VERSION':'v1', # 默認版本
}
#若是加版本url要以下配置(?P<version>\w+)，若是繼承的是APIview,方法中要加*args,**kwargs，
# 若是不加版本就不用加*args,**kwargs，獲取版本用request.version,
#url(r'^(?P<version>\w+)/book/$', views.BookView.as_view()),request.version,獲取版本

#url(r'^(?P<version>[v1|v2]+)/users/$', users_list, name='users-list'),這種只能寫v1/v2，因此用w+

urls.pyjson

from django.conf.urls import url
from django.contrib import admin
from app01 import views

urlpatterns = [
    url(r'^admin/', admin.site.urls),
    url(r'^publisher/$', views.PublisherView.as_view()),
    url(r'^publisher/(\d+)/$', views.PublisherDetailView.as_view()),

    # url(r'^book/$', views.BookView.as_view()),
    # url(r'^book/(?P<pk>\d+)/$', views.BookDetailView.as_view()),
    #加版本
    url(r'^(?P<version>\w+)/book/$', views.BookView.as_view()),
    url(r'^(?P<version>\w+)/book/(?P<pk>\d+)/$', views.BookDetailView.as_view()),


    url(r'^books/$', views.BookViewSet.as_view({"get":"list","post":"create"}),name="book_list"),
    url(r'^books/(?P<pk>\d+)/$', views.BookViewSet.as_view({
                'get': 'retrieve',
                'put': 'update',
                'patch': 'partial_update',
                'delete': 'destroy'
            }),name="book_detail"),

    url(r'^auth/$', views.AuthView.as_view()),
]

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。