rest_framework自己總結的
阿新 • • 發佈:2018-11-14
settings.py
REST_FRAMEWORK = { "DEFAULT_VERSIONING_CLASS": "rest_framework.versioning.URLPathVersioning", "DEFAULT_VERSION": "v1", "ALLOWED_VERSIONS": ["v1", "v2"], "VERSION_PARAM": "version", # "DEFAULT_AUTHENTICATION_CLASSES": ["utils.Auth.MyAuth"], "DEFAULT_THROTTLE_RATES": { "WD": "3/m" } }
urls.py
from django.conf.urls import url from django.contrib import admin # from serializers.views import StudentAPIView, StudentEditView, BooksAPIView from serializers.views import StudentView, VersionView, UserView, TestAuthView,BookPageView urlpatterns = [ url(r'^admin/', admin.site.urls), # url(r'^api/students', StudentAPIView.as_view()), # get post請求 # url(r'^api/student/(?P<id>\d+)', StudentEditView.as_view()), #帶有id的get patch delete請求 # 使用我們自己寫的ModelViewSet,繼承ViewSetMixin進行請求的分發 # url(r'^api/students', BooksAPIView.as_view({"get": "list", "post": "create"})), # url(r'^api/student/(?P<id>\d+)', BooksAPIView.as_view({"get": "retrieve", "patch": "update", "delete": "destroy"})), # 使用rest_framework自帶的ModelViewSet進行請求的分發,這裡要注意url攜帶的命名引數,名字應該是pk # url(r'^api/students', BooksAPIView.as_view({"get": "list", "post": "create"})), # url(r'^api/student/(?P<pk>\d+)', BooksAPIView.as_view({"get": "retrieve", "patch": "update", "delete": "destroy"})), # 自己寫的測試版二,可以通用的版本 # url(r'^api/student$', StudentView.as_view()), # url(r'^api/student/(?P<id>\d+)', StudentView.as_view()), # 帶版本控制的 url(r'^(?P<version>[v1|v2]+)/api/student$', StudentView.as_view()), # http://127.0.0.1:8000/v1/api/student url(r'^(?P<version>[v1|v2]+)/api/student/(?P<id>\d+)', StudentView.as_view()), # 版本控制測試 url(r'^(?P<version>[v1|v2]+)/book$', VersionView.as_view()), # 認證測試,許可權測試,頻率測試 url(r'^user$', UserView.as_view()), url(r'^test', TestAuthView.as_view()), # http://127.0.0.1:8000/test?token=20609ddd01fe4faeb0ffe7d8d8c39881 # 分頁測試 url(r'^book_page', BookPageView.as_view()), # http://127.0.0.1:8000/book_page?page=1&size=1 # http://127.0.0.1:8000/book_page?limit=2&offset=0 # http://127.0.0.1:8000/book_page ]
utils資料夾下的檔案 Auth.py(認證) pagenation.py(分頁) permissions.py(許可權) throttle(頻率)
# Auth.py from rest_framework.authentication import BaseAuthentication from rest_framework.exceptions import AuthenticationFailed from serializers.models import UserInfo from rest_framework.response import Response class MyAuth(BaseAuthentication): def authenticate(self, request): # 第一步先拿到前端傳過來的token token = request.query_params["token"] # 驗證token是否存在 user_obj = UserInfo.objects.filter(token=token).first() if user_obj: return (user_obj, token) else: raise AuthenticationFailed("認證失敗") # pagenation.py from rest_framework import pagination # 分頁 class MyPagenation(pagination.PageNumberPagination): page_size = 2 page_query_param = 'page' page_size_query_param = "size" max_page_size = 3 class MyLimitPage(pagination.LimitOffsetPagination): default_limit = 1 limit_query_param = 'limit' # limit限制每頁顯示的個數 offset_query_param = 'offset' # 相對第一個資料的偏移量 max_limit = 2 class MyCursorPage(pagination.CursorPagination): cursor_query_param = 'cursor' page_size = 2 ordering = '-id' # permissions.py class MyPermission(object): message = "您沒有許可權,請充值" def has_permission(self, request, view): # 許可權邏輯 有許可權返回True 沒有返回False # 認證是在許可權前面執行 # request.user user_obj user_obj = request.user if user_obj.type == 1: return True else: return False # throttle.py from rest_framework import throttling # 頻率 import time VISIT_RECORD = {} class MyThrottle(object): """ 60秒訪問3次 """ def __init__(self): self.history = None def allow_request(self, request, view): """ 頻率限制的邏輯 通過返回True 不通過返回False :param request: :param view: :return: """ # 獲取使用者IP ip = request.META.get("REMOTE_ADDR") # 判斷ip是否在訪問記錄裡 now = time.time() if ip not in VISIT_RECORD: VISIT_RECORD[ip] = [now,] # 如果ip在訪問記錄裡 history = VISIT_RECORD[ip] # 把當然訪問時間新增到列表最前面 history.insert(0, now) # 確保列表內的時間都是範圍內時間 while history and now - history[-1] > 60: history.pop() self.history = history # 看列表長度是否符合限制次數 if len(history) <= 3: # 經過這樣的限制後,history列表中(60秒內)最多有三次訪問記錄,列表中最後的資料是據現在60秒內最早的訪問記錄(訪問時間) return True else: return False def wait(self): """ 返回還剩多久可以訪問 :return: """ now = time.time() return 60 - (now - self.history[-1]) class MyVisitThrottle(throttling.SimpleRateThrottle): scope = "WD" """ 第一 自己的類裡要有scope 第二 settings DEFAULT_THROTTLE_RATES 第三 DEFAULT_THROTTLE_RATES = { scope配置的變數值:xxx} 第四 重寫 get_cache_key(self, request, view) """ def get_cache_key(self, request, view): return self.get_ident(request)
views.py
from django.shortcuts import render from rest_framework.views import APIView from rest_framework.response import Response from .models import Student,UserInfo from .serializers import StudentSerializer from utils.Auth import MyAuth from utils.permissions import MyPermission from utils.throttle import MyThrottle, MyVisitThrottle import uuid from utils.pagenation import MyPagenation,MyLimitPage,MyCursorPage from rest_framework.viewsets import ViewSetMixin, ModelViewSet # Create your views here. """ class GenericAPIView(APIView): queryset = None serializer_class = None def get_queryset(self): return self.queryset.all() def get_serializer(self, *args, **kwargs): return self.serializer_class(*args, **kwargs) class ListModelMixin(object): def list(self, request): queryset = self.get_queryset() ser_obj = self.get_serializer(queryset, many=True) return Response(ser_obj.data) class CreateModelMixin(object): def create(self, request): ser_obj = self.get_serializer(data=request.data) if ser_obj.is_valid(): print(ser_obj.validated_data) ser_obj.save() return Response(ser_obj.validated_data) else: return Response(ser_obj.errors) class All(GenericAPIView, ListModelMixin, CreateModelMixin): pass class RetrieveModelMixin(object): def retrieve(self, request, id): student_obj = self.get_queryset().filter(id=id).first() ser_obj = self.get_serializer(student_obj) return Response(ser_obj.data) class UpdateModelMixin(object): def update(self, request, id): student_obj = self.get_queryset().filter(id=id).first() ser_obj = self.get_serializer(instance=student_obj, data=request.data, partial=True) if ser_obj.is_valid(): ser_obj.save() return Response(ser_obj.validated_data) else: return Response(ser_obj.errors) class DestroyModelMixin(object): def destroy(self, request, id): student_obj = self.get_queryset().filter(id=id).first() if student_obj: student_obj.delete() return Response("") else: return Response("刪除物件不存在") class EditAll(GenericAPIView, RetrieveModelMixin, UpdateModelMixin, DestroyModelMixin): pass class StudentAPIView(All): queryset = Student.objects.all() serializer_class = StudentSerializer def get(self, request): return self.list(request) def post(self, request): return self.create(request) class StudentEditView(EditAll): queryset = Student.objects.all() serializer_class = StudentSerializer def get(self, request, id): # 不知道為什麼這裡的形參id的名字必須與url(?P<id>\d+)中傳遞引數的名字一樣,可能這裡傳的是關鍵字引數吧 return self.retrieve(request, id) def patch(self, request, id): return self.update(request, id) def delete(self, request, id): return self.destroy(request, id) # class ModelViewSet(ViewSetMixin, All, EditAll): # 開啟註釋用的是自己寫的ModelViewSet,不開啟用的是rest_framework封裝好的ModelViewSet # pass class BooksAPIView(ModelViewSet): queryset = Student.objects.all() serializer_class = StudentSerializer """ """ # 自己的測試版一 class GenericAPIView(APIView): queryset = None serializer_class = None def get_queryset(self): return self.queryset.all() def get_serializer(self, *args, **kwargs): return self.serializer_class(*args, **kwargs) class RetrieveModelMixin(object): def retrieve(self, request, id): if not id: queryset = self.get_queryset() ser_obj = self.get_serializer(queryset, many=True) else: student_obj = self.get_queryset().filter(id=id).first() ser_obj = self.get_serializer(student_obj) return Response(ser_obj.data) class CreateModelMixin(object): def create(self, request): ser_obj = self.get_serializer(data=request.data) if ser_obj.is_valid(): print(ser_obj.validated_data) ser_obj.save() return Response(ser_obj.validated_data) else: return Response(ser_obj.errors) class UpdateModelMixin(object): def update(self, request, id): student_obj = self.get_queryset().filter(id=id).first() ser_obj = self.get_serializer(instance=student_obj, data=request.data, partial=True) if ser_obj.is_valid(): ser_obj.save() return Response(ser_obj.validated_data) else: return Response(ser_obj.errors) class DestroyModelMixin(object): def destroy(self, request, id): student_obj = self.get_queryset().filter(id=id).first() if student_obj: student_obj.delete() return Response("") else: return Response("刪除物件不存在") class All(GenericAPIView,RetrieveModelMixin,CreateModelMixin,UpdateModelMixin,DestroyModelMixin): pass class StudentView(All): queryset = Student.objects.all() serializer_class = StudentSerializer def get(self, request, id=None): return self.retrieve(request, id) def post(self, request,id=None): return self.create(request) def patch(self, request, id=None): return self.update(request, id) def delete(self, request, id=None): return self.destroy(request, id) """ """ # 自己的測試版二,這版整理的比較好可以通用 class GenericAPIView(APIView): queryset = None serializer_class = None def get_queryset(self): return self.queryset.all() def get_serializer(self, *args, **kwargs): return self.serializer_class(*args, **kwargs) def retrieve(self, request, id): if not id: queryset = self.get_queryset() ser_obj = self.get_serializer(queryset, many=True) else: student_obj = self.get_queryset().filter(id=id).first() ser_obj = self.get_serializer(student_obj) return Response(ser_obj.data) def create(self, request): ser_obj = self.get_serializer(data=request.data) if ser_obj.is_valid(): print(ser_obj.validated_data) ser_obj.save() return Response(ser_obj.validated_data) else: return Response(ser_obj.errors) def update(self, request, id): student_obj = self.get_queryset().filter(id=id).first() ser_obj = self.get_serializer(instance=student_obj, data=request.data, partial=True) if ser_obj.is_valid(): ser_obj.save() return Response(ser_obj.validated_data) else: return Response(ser_obj.errors) def destroy(self, request, id): student_obj = self.get_queryset().filter(id=id).first() if student_obj: student_obj.delete() return Response("") else: return Response("刪除物件不存在") class StudentView(GenericAPIView): queryset = Student.objects.all() serializer_class = StudentSerializer def get(self, request, id=None): return self.retrieve(request, id) def post(self, request,id=None): return self.create(request) def patch(self, request, id=None): return self.update(request, id) def delete(self, request,id=None): return self.destroy(request, id) """ # url帶版本驗證 class GenericAPIView(APIView): queryset = None serializer_class = None def get_queryset(self): return self.queryset.all() def get_serializer(self, *args, **kwargs): return self.serializer_class(*args, **kwargs) def retrieve(self, request,version, id): if not id: queryset = self.get_queryset() ser_obj = self.get_serializer(queryset, many=True) else: student_obj = self.get_queryset().filter(id=id).first() ser_obj = self.get_serializer(student_obj) return Response(ser_obj.data) def create(self, request): ser_obj = self.get_serializer(data=request.data) if ser_obj.is_valid(): print(ser_obj.validated_data) ser_obj.save() return Response(ser_obj.validated_data) else: return Response(ser_obj.errors) def update(self, request, id): student_obj = self.get_queryset().filter(id=id).first() ser_obj = self.get_serializer(instance=student_obj, data=request.data, partial=True) if ser_obj.is_valid(): ser_obj.save() return Response(ser_obj.validated_data) else: return Response(ser_obj.errors) def destroy(self, request, id): student_obj = self.get_queryset().filter(id=id).first() if student_obj: student_obj.delete() return Response("") else: return Response("刪除物件不存在") class StudentView(GenericAPIView): queryset = Student.objects.all() serializer_class = StudentSerializer def get(self, request,version=None, id=None): # 這裡做些版本控制的邏輯,不同版本請求做不同處理,當然這裡的邏輯最好還是寫到GenericAPIView類裡面 if request.version == "v1": return Response("v1版本的返回") elif request.version == "v2": return Response("v2版本的返回") return self.retrieve(request,version, id) def post(self, request,version=None,id=None): return self.create(request) def patch(self, request,version=None, id=None): return self.update(request, id) def delete(self, request,version=None, id=None): return self.destroy(request, id) class VersionView(APIView): def get(self, request, version): print(request.version) print(request.versioning_scheme) if request.version == "v1": return Response("v1版本的返回") elif request.version == "v2": return Response("v2版本的返回") return Response("版本不存在") class UserView(APIView): def post(self, request): # 這相當於註冊 username = request.data["username"] UserInfo.objects.create(username=username, token=uuid.uuid4()) return Response("ok") class TestAuthView(APIView): authentication_classes = [MyAuth,] # 這裡是區域性註冊認證,寫在settings中是全域性註冊 permission_classes = [MyPermission, ] throttle_classes = [MyThrottle, ] # 使用自己寫的 # throttle_classes = [MyVisitThrottle, ] # 使用提供的頻率控制 # 這相當於登入的認證 def get(self, request): print(request.user) print(request.auth) return Response("認證測試") from rest_framework.negotiation import DefaultContentNegotiation class BookPageView(APIView): def get(self, request): queryset = Student.objects.all() my_pagenation = MyPagenation() # page size分頁方法 # my_pagenation = MyLimitPage() # limit offset分頁方法 # my_pagenation = MyCursorPage() # cursor 遊標分頁 page_queryset = my_pagenation.paginate_queryset(queryset, request, view=self) ser_obj = StudentSerializer(page_queryset, many=True) # return Response(ser_obj.data) return my_pagenation.get_paginated_response(ser_obj.data) def post(self, request): ser_obj = StudentSerializer(data=request.data) if ser_obj.is_valid(): ser_obj.save() return Response(ser_obj.validated_data) else: return Response(ser_obj.errors)
serializers.py
from rest_framework import serializers from .models import Student class ClassesSerializer(serializers.Serializer): id = serializers.IntegerField() name = serializers.CharField(max_length=32) class CourseSerializer(serializers.Serializer): id = serializers.IntegerField() name = serializers.CharField(max_length=32) class StudentSerializer(serializers.ModelSerializer): classes_info = serializers.SerializerMethodField(read_only=True) course_info = serializers.SerializerMethodField(read_only=True) def get_classes_info(self, obj): ret = { "id": obj.classes.id, "name": obj.classes.name } return ret def get_course_info(self, obj): courses = obj.courses.all() ret = [] for course in courses: ret.append({ "id": course.id, 'name': course.name }) return ret class Meta: model = Student fields = "__all__" # fields = ["id", "name", "classes_info","course_info"] extra_kwargs = { "classes": {"write_only": True}, "courses": {'write_only': True}, }
models.py
from django.db import models # Create your models here. __all__ = ['Student', 'Classes', 'Course'] class Student(models.Model): name = models.CharField(max_length=32, verbose_name='姓名') age = models.IntegerField() classes = models.ForeignKey(to='Classes') courses = models.ManyToManyField(to='Course') def __str__(self): return self.name class Meta: db_table='01-學生表' verbose_name_plural=db_table class Classes(models.Model): name = models.CharField(max_length=32,verbose_name='班級') def __str__(self): return self.name class Meta: db_table='02-班級表' verbose_name_plural=db_table class Course(models.Model): name=models.CharField(max_length=32,verbose_name='課程') def __str__(self): return self.name class Meta: db_table='03-課程表' verbose_name_plural=db_table class UserInfo(models.Model): username = models.CharField(max_length=32) token = models.UUIDField() CHOICES = ((1, "vip"), (2, "svip"), (3, "普通使用者")) type = models.IntegerField(choices=CHOICES, default=3)
app下的admin.py
from django.contrib import admin from . import models # Register your models here. for table in models.__all__: admin.site.register(getattr(models,table))