本文将介绍DjangoRestFramework部分更新的详细情况,特别是关于django局部更新数据的相关信息。我们将通过案例分析、数据研究等多种方式,帮助您更全面地了解这个主题,同时也将涉及一些关
本文将介绍Django Rest Framework部分更新的详细情况,特别是关于django 局部更新数据的相关信息。我们将通过案例分析、数据研究等多种方式,帮助您更全面地了解这个主题,同时也将涉及一些关于1 Django REST Framework 开发 ---- 原生 Django View 热身、Django - rest - framework - 下、Django Rest Framework、Django rest framework (4) --- 版本的知识。
本文目录一览:- Django Rest Framework部分更新(django 局部更新数据)
- 1 Django REST Framework 开发 ---- 原生 Django View 热身
- Django - rest - framework - 下
- Django Rest Framework
- Django rest framework (4) --- 版本
Django Rest Framework部分更新(django 局部更新数据)
我正在尝试partial_update
使用Django Rest Framework进行实施,但由于我被卡住了,因此需要澄清一下。
为什么我们需要指定partial = True?
以我的理解,我们可以轻松地在partial_update
方法内部更新Demo对象。这样做的目的是什么?序列化变量里面有什么? 方法变量
内部是什么?那是一个演示对象吗?在后台调用什么功能?serialized``partial_update
一个人如何在这里完成实施?
视图集
class DemoViewSet(viewsets.ModelViewSet): serializer_class = DemoSerializer def partial_update(self, request, pk=None): serialized = DemoSerializer(request.user, data=request.data, partial=True) return Response(status=status.HTTP_202_ACCEPTED)
序列化器
class DemoSerializer(serializers.ModelSerializer): class Meta: model = Demo fields = ''__all__'' def update(self, instance, validated_data): print ''this - here'' demo = Demo.objects.get(pk=instance.id) Demo.objects.filter(pk=instance.id)\ .update(**validated_data) return demo
答案1
小编典典我和您以前有相同的问题,但是当我深入研究rest_framework的源代码时,我得到了以下发现,希望对您有所帮助:
对于问题1。为什么我们需要指定partial = True?
这个问题与HTTP动词有关。
PUT :PUT方法用请求有效载荷替换目标资源的所有当前表示形式。
PATCH :PATCH方法用于对资源进行部分修改。
通常,partial
用于在客户向视图提交数据时检查模型中的字段是否需要进行字段验证。
例如,我们有一个Book
这样的模型,请注意name
和author_name
字段都是 必填 字段(不为null且不为空白)。
class Book(models.Model): name = models.CharField(''name of the book'', max_length=100) author_name = models.CharField(''the name of the author'', max_length=50)# Create a new instance for testingBook.objects.create(name=''Python in a nut shell'', author_name=''Alex Martelli'')
在某些情况下,我们可能只需要更新模型中的部分字段,例如,我们只需要更新中的name
字段即可Book
。因此,在这种情况下,客户端将只向name
视图提交具有新值的字段。从客户端提交的数据可能如下所示:
{"pk": 1, name: "PYTHON IN A NUT SHELL"}
但是您可能已经注意到,我们的模型定义不允许author_name
为空。因此,我们必须使用 partial_update
而不是update
。因此,其余框架将不会对请求数据中缺少的字段执行 字段验证检查 。
出于测试目的,您可以为update
和创建两个视图partial_update
,您将对我刚才所说的内容有更多的了解。
例:
views.py
from rest_framework.generics import GenericAPIViewfrom rest_framework.mixins import UpdateModelMixinfrom rest_framework.viewsets import ModelViewSetfrom rest_framework import serializersclass BookSerializer(serializers.ModelSerializer): class Meta: model = Bookclass BookUpdateView(GenericAPIView, UpdateModelMixin): '''''' Book update API, need to submit both `name` and `author_name` fields At the same time, or django will prevent to do update for field missing '''''' queryset = Book.objects.all() serializer_class = BookSerializer def put(self, request, *args, **kwargs): return self.update(request, *args, **kwargs)class BookPartialUpdateView(GenericAPIView, UpdateModelMixin): '''''' You just need to provide the field which is to be modified. '''''' queryset = Book.objects.all() serializer_class = BookSerializer def put(self, request, *args, **kwargs): return self.partial_update(request, *args, **kwargs)
urls.py
urlpatterns = patterns('''', url(r''^book/update/(?P<pk>\d+)/$'', BookUpdateView.as_view(), name=''book_update''), url(r''^book/update-partial/(?P<pk>\d+)/$'', BookPartialUpdateView.as_view(), name=''book_partial_update''),)
提交数据
{"pk": 1, name: "PYTHON IN A NUT SHELL"}
当您将上述json提交到时/book/update/1/
,您会收到HTTP_STATUS_CODE = 400的以下错误:
{ "author_name": [ "This field is required." ]}
但是,当您将上述json提交给时/book/update-partial/1/
,您将获得HTTP_STATUS_CODE = 200并显示以下响应,
{ "id": 1, "name": "PYTHON IN A NUT SHELL", "author_name": "Alex Martelli"}
对于问题2。序列化变量的内部是什么?
serialized
是将模型实例包装为可序列化对象的对象。并且您可以使用此序列化生成带有的纯JSON字符串serialized.data
。
对于问题3。如何在这里完成实施?
我认为您在阅读完上述答案后就可以回答自己,并且应该知道何时使用update
和何时使用partial_update
。
如果您还有任何疑问,请随时提问。我只是阅读了rest框架的部分源代码,可能对某些术语不太了解,请在错误的地方指出来…
1 Django REST Framework 开发 ---- 原生 Django View 热身
在正式开始 Django REST Framework 学习之前,先用 django.core.serializers 和原生的 Django View 类来实现一次数据的序列化数据。
下面的例子展示了,是如何解决序列化 JSON 数据的,前提:给Products的Model已经创建完
创建 Django 项目,并修改 urls.py 文件:
1 from django.views.generic import TemplateView
2 from product.views import ProductsListView
3
4
5 urlpatterns = [
6 url(r''^admin/'', admin.site.urls),
7 url(r''^products/'', ProductsListView.as_view(), name="product-list"),
8 ]
创建 view.py 文件并写入如下内容:
1 import json
2
3 from django.http import JsonResponse
4 from django.core import serializers
5 from django.views.generic.base import View
6
7 from products.models import Product
8
9
10 class ProductListView(View):
11 def get(self, request):
12 products = Product.objects.all()
13 json_data = serializers.serialize(''json'', products) # 将django model 得到的对象序列化成 json 类型
14 json_data = json.loads(json_data)
15
16 return JsonResponse(json_data, safe=False) # safe = False 因为json_data是 non-dict 类型,不然会报错
注:在打开网页显示数据时需下载chrome jsonview插件
这样一个基本的django 使用json数据类型调取数据的过程就完成了,但是会有以下问题:
- 图片,日期格式支持不够(仅会显示图片相对路径)
- 主键JSON字段之外,使得调用起来会不方便
- 文档需要手动写,缺乏规范性且效率低下
- 输入验证缺失 (类似于表单的认证)
- 等等
Django - rest - framework - 下
一、视图三部曲
https://www.cnblogs.com/wupeiqi/articles/7805382.html
使用混合(mixins)
之前得视图部分
# urls.py
from django.conf.urls import url
from django.contrib import admin
from app01 import views
urlpatterns = [
url(r''^admin/'', admin.site.urls),
url(r''^publishes/$'', views.PublishView.as_view(),name="publish"),
url(r''^publishes/(?P<pk>\d+)/$'', views.PublishDetailView.as_view(),name="detail_publish"),
url(r"^books/$", views.BookView.as_view(),name="books"),
url(r''^books/(?P<pk>\d+)/$'',views.BookDetailView.as_view(),name="detail_book")
]
------------------------------------------------------
# views.py
from rest_framework.views import APIView, Response
from app01.serializers import *
class PublishView(APIView):
def get(self, request):
publish_list = Publish.objects.all()
ret = PublishModelSerializers(publish_list, many=True)
return Response(ret.data)
def post(self, request):
ps = PublishModelSerializers(data=request.data)
if ps.is_valid():
ps.save()
return Response(ps.data)
else:
return Response(ps.errors)
class PublishDetailView(APIView):
def get(self,request,pk):
publish = Publish.objects.filter(pk=pk).first()
ps = PublishModelSerializers(publish)
return Response(ps.data)
def put(self,request,pk):
publish = Publish.objects.filter(pk=pk).first()
ps = PublishModelSerializers(publish,data=request.data)
if ps.is_valid():
ps.save()
return Response(ps.data)
else:
return Response(ps.errors)
def delete(self,request,pk):
Publish.objects.filter(pk=pk).delete()
return Response()
class BookView(APIView):
def get(self, request):
book_list = Book.objects.all()
ret = BookModelSerializers(book_list, many=True,context={"request":request})
return Response(ret.data)
def post(self, request):
bms = BookModelSerializers(data=request.data, many=False,context={"request":request})
if bms.is_valid():
bms.save()
return Response(bms.data)
else:
return Response(bms.errors)
class BookDetailView(APIView):
def get(self,request,pk):
book = Book.objects.filter(pk=pk).first()
# 序列化
bms = BookModelSerializers(book,context={"request":request})
return Response(bms.data)
def put(self,request,pk):
book = Book.objects.filter(pk=pk).first()
bms = BookModelSerializers(book,data=request.data,context={"request":request})
if bms.is_valid():
bms.save()
return Response(bms.data)
else:
return Response(bms.errors)
def delete(self,reqeust,pk):
Book.objects.filter(pk=pk).delete()
return Response()
---------------------------------------------------
# serializers.py
# -*- coding:utf-8 -*-
from .models import *
from rest_framework import serializers
class PublishModelSerializers(serializers.ModelSerializer):
class Meta:
model = Publish
fields = "__all__"
class BookModelSerializers(serializers.ModelSerializer):
class Meta:
model = Book
fields = "__all__"


from django.db import models
# Create your models here.
class Book(models.Model):
title = models.CharField(max_length=32)
price = models.IntegerField()
pub_date = models.DateField()
publish = models.ForeignKey("Publish")
authors = models.ManyToManyField("Author")
def __str__(self):
return self.title
class Publish(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
1. mixin类编写视图
from rest_framework import mixins, generics
mixins.ListModelMixin, mixins.CreateModelMixin,
mixins.RetrieveModelMixin, mixins.UpdateModelMixin, mixins.DestroyModelMixin,
generics.GenericAPIView
url(r''^authors/$'',views.AuthorView.as_view(),name="author"),
url(r''^authors/(?P<pk>\d+)'',views.AuthorDetailView.as_view(),name="detail_author"),
-----------------------------------------
######################### mixin类编写视图 ##############################
from rest_framework import mixins, generics
class AuthorView(mixins.ListModelMixin, mixins.CreateModelMixin, generics.GenericAPIView):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
def get(self, request, *args, **kwargs):
return self.list(request, *args, **kwargs)
def post(self, request, *args, **kwargs):
return self.create(request, *args, **kwargs)
class AuthorDetailView(mixins.RetrieveModelMixin, mixins.UpdateModelMixin, mixins.DestroyModelMixin,
generics.GenericAPIView):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
def get(self, request, *args, **kwargs):
return self.retrieve(request, *args, **kwargs)
def put(self, request, *args, **kwargs):
return self.update(request, *args, **kwargs)
def delete(self, request, *args, **kwargs):
return self.destroy(request, *args, **kwargs)
------------------------------------------------
class AuthorModelSerializers(serializers.ModelSerializer):
class Meta:
model = Author
fields = "__all__"
2. 使用通用的基于类的视图
from rest_framework import generics
generics.ListCreateAPIView
generics.RetrieveUpdateDestroyAPIView
通过使用mixin类,我们使用更少的代码重写了这些视图,但我们还可以再进一步。REST框架提供了一组已经混合好(mixed-in)的通用视图,我们可以使用它来简化我们的views.py
模块。
######################### 使用通用得基于类得视图 ##############################
from rest_framework import generics
class AuthorView(generics.ListCreateAPIView):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
class AuthorDetailView(generics.RetrieveUpdateDestroyAPIView):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
3. viewsets.ModelViewSet
views.AuthorModelView.as_view({"get": "list", "post": "create"})
views.AuthorModelView.as_view({"get": "retrieve", "put": "update", "delete": "destroy"})
from rest_framework import viewsets
viewsets.ModelViewSet
url(r''^authors/$'', views.AuthorModelView.as_view({"get": "list", "post": "create"}), name="author"),
url(r''^authors/(?P<pk>\d+)'',
views.AuthorModelView.as_view({"get": "retrieve", "put": "update", "delete": "destroy"}), name="detail_author"),
---------------------------------------------------
######################### viewsets.ModelViewSet ##############################
from rest_framework import viewsets
class AuthorModelView(viewsets.ModelViewSet):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
# 可重写,覆盖!
# def list(self,request,*args,**kwargs):pass
效果:
get post get put delete 都可访问!


http://www.cnblogs.com/yuanchenqi/articles/8719520.html
视图三部曲
5中方法: 查(全部) 查(单条) 增 删 改
逻辑封装起来了
-----------------------------------
class AuthorModelView(viewsets.ModelViewSet):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
(1):url(r''^authors/$'', views.AuthorModelView.as_view({"get":"list","post":"create"}), name=''author''),
(2):url(r''^authors/$'', ViewSetMixin.as_view({"get":"list","post":"create"}), name=''author''),
(3):url(r''^authors/$'', ViewsetMixin.View, name=''author''),
一旦用户 get 方式 访问 authors:
ViewsetMixin.View():
for method, action in actions.items(): # {"get":"list","post":"create"}
handler = getattr(self, action) # self.list self.create
setattr(self, method, handler) # self.get = self.list self.post = self.create
# getattr(self,"get") # self.list
# getattr(self,"post") # self.create
return self.dispatch()
APIView.dispatch():
if request.method.lower() in self.http_method_names:
handler = getattr(self,request.method.lower())
response = handler(request,*args,**kwargs) # self.list()
return response
(ViewSetMixin)
二、认证组件
生成随机字符串
import hashlib, time
def get_random_str(user):
""" 生成随机 字符串 """
ctime = str(time.time())
md5 = hashlib.md5(bytes(user,encoding=''utf-8''))
md5.update(bytes(ctime,encoding="utf-8"))
return md5.hexdigest()
update_or_create
# update_or_create
Token.objects.update_or_create(user=user,defaults={"token":random_str})
返回json
# import json
# from django.shortcuts import HttpResponse
# return HttpResponse(json.dumps(res,ensure_ascii=False))
# from django.http import JsonResponse
# return JsonResponse(res)
return Response(res)
登录,生成随机token
url(r''^login/$'', views.LoginView.as_view(), name="login")
---------------------------------------------
import hashlib, time
def get_random_str(user):
""" 生成随机 字符串 """
ctime = str(time.time())
md5 = hashlib.md5(bytes(user,encoding=''utf-8''))
md5.update(bytes(ctime,encoding="utf-8"))
return md5.hexdigest()
class LoginView(APIView):
def post(self,request):
name = request.data.get("name")
pwd = request.data.get("pwd")
user = User.objects.filter(name=name,pwd=pwd).first()
res = {"state_code":1000,"msg":None}
if user:
random_str = get_random_str(user.name)
# update_or_create
Token.objects.update_or_create(user=user,defaults={"token":random_str})
res["token"] = random_str
else:
res["state_code"] = 1001 # 错误状态码
res["msg"] = "用户名或密码错误"
# import json
# from django.shortcuts import HttpResponse
# return HttpResponse(json.dumps(res,ensure_ascii=False))
# from django.http import JsonResponse
# return JsonResponse(res)
return Response(res)
--------------------------------------------------------
# models.py
class User(models.Model):
name = models.CharField(max_length=32)
pwd = models.CharField(max_length=32)
class Token(models.Model):
user = models.OneToOneField("User")
token = models.CharField(max_length=128)
def __str__(self):
return self.token
效果图:
登录验证 - 局部
authentication_classes = [TokenAuth]
from rest_framework import exceptions
from rest_framework.authentication import BaseAuthentication
def authenticate(self,request):
... ...
if not token_obj:
raise exceptions.AuthenticationFailed("验证失败")
return (token_obj.user.name, token_obj)
from rest_framework import exceptions
# class TokenAuth(object):
# def authenticate(self,request):
# token = request.GET.get("token")
# token_obj = Token.objects.filter(token=token).first()
# if not token_obj:
# raise exceptions.AuthenticationFailed("验证失败")
#
# return (token_obj.user.name, token_obj)
#
# def authenticate_header(self,request):
# pass
from rest_framework.authentication import BaseAuthentication
class TokenAuth(BaseAuthentication):
def authenticate(self,request):
token = request.GET.get("token")
token_obj = Token.objects.filter(token=token).first()
if not token_obj:
raise exceptions.AuthenticationFailed("验证失败")
return (token_obj.user.name, token_obj)
class BookView(APIView):
authentication_classes = [TokenAuth]
def get(self, request):
book_list = Book.objects.all()
ret = BookModelSerializers(book_list, many=True, context={"request": request})
return Response(ret.data)
def post(self, request):
bms = BookModelSerializers(data=request.data, many=False, context={"request": request})
if bms.is_valid():
bms.save()
return Response(bms.data)
else:
return Response(bms.errors)
登录验证 - 全局
settings 配置
''DEFAULT_AUTHENTICATION_CLASSES'': [''app01.utils.TokenAuth'']
# settings.py
REST_FRAMEWORK = {
''DEFAULT_AUTHENTICATION_CLASSES'': [''app01.utils.TokenAuth'']
}
-------------------------------------------
# app01.utils
from .models import *
from rest_framework import exceptions
from rest_framework.authentication import BaseAuthentication
class TokenAuth(BaseAuthentication):
def authenticate(self,request):
token = request.GET.get("token")
token_obj = Token.objects.filter(token=token).first()
if not token_obj:
raise exceptions.AuthenticationFailed("验证失败")
return (token_obj.user.name, token_obj)
--------------------------------------------
# views.py
class BookView(APIView):
def get(self, request):
book_list = Book.objects.all()
ret = BookModelSerializers(book_list, many=True, context={"request": request})
return Response(ret.data)
def post(self, request):
bms = BookModelSerializers(data=request.data, many=False, context={"request": request})
if bms.is_valid():
bms.save()
return Response(bms.data)
else:
return Response(bms.errors)
-----------------------------------------
class AuthorModelView(viewsets.ModelViewSet):
authentication_classes = [] # 加上这个,前提是全局有认证;加上这个,就走自己得,不认证了,自己没有,才走全局配置得!
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
效果图:
三、权限组件
权限 - 局部
permission_classes = [SVIPPermission]
has_permission(self,request,view):pass # 固定得写法,根据源码来写得!
authentication_classes = [] # 登录页面 不需要验证,在全局配置得前提下
class AuthorModelView(viewsets.ModelViewSet):
authentication_classes = [TokenAuth]
permission_classes = [SVIPPermission]
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
----------------------------------------------
# app01.utils.SVIPPermission
class SVIPPermission(object):
message = "只有超级用户才能访问"
def has_permission(self,request,view):
username = request.user
user_type = User.objects.filter(name=username).first().user_type
if user_type == 3:
return True
else:
return False
----------------------------------------------
class LoginView(APIView):
authentication_classes = []
...
...
权限 - 全局
settings配置:
''DEFAULT_PERMISSION_CLASSES'': [''app01.utils.SVIPPermission'']
# settings.py
REST_FRAMEWORK = {
''DEFAULT_AUTHENTICATION_CLASSES'': [''app01.utils.TokenAuth''],
''DEFAULT_PERMISSION_CLASSES'': [''app01.utils.SVIPPermission'']
}
效果图:
四、频率组件
局部视图throttle
class BookView(APIView):
# authentication_classes = [TokenAuth]
# permission_classes = [SVIPPermission]
throttle_classes = [VisitRateThrottle]
。。。 。。。
-----------------------------------------------
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)<3:
history.insert(0,ctime)
return True
else:
return False
def wait(self):
import time
ctime=time.time()
return 60-(ctime-self.history[-1])
全局视图throttle
REST_FRAMEWORK = {
''DEFAULT_AUTHENTICATION_CLASSES'': [''app01.utils.TokenAuth''],
''DEFAULT_PERMISSION_CLASSES'': [''app01.utils.SVIPPermission''],
''DEFAULT_THROTTLE_CLASSES'': [''app01.utils.VisitThrottle''],
}
内置throttle类
REST_FRAMEWORK = {
''DEFAULT_AUTHENTICATION_CLASSES'': [''app01.utils.TokenAuth''],
''DEFAULT_PERMISSION_CLASSES'': [''app01.utils.SVIPPermission''],
''DEFAULT_THROTTLE_CLASSES'': [''app01.utils.VisitThrottle''],
"DEFAULT_THROTTLE_RATES": {
"visit_rate": "1/m",
}
}
------------------------------------
from rest_framework.throttling import SimpleRateThrottle
class VisitThrottle(SimpleRateThrottle):
scope="visit_rate"
def get_cache_key(self, request, view):
return self.get_ident(request)
http://www.cnblogs.com/yuanchenqi/articles/8719520.html
五、解析器
from rest_framework.parsers import JSONParser,FormParser,MultiPartParser,FileUploadParser
"""
默认得是 JSONParser FormParser MultiPartParser
"""
class BookView(APIView):
parser_classes = [JSONParser,FormParser]
...
request类
django的request类和rest-framework的request类的源码解析
局部视图
from rest_framework.parsers import JSONParser,FormParser
class PublishViewSet(generics.ListCreateAPIView):
parser_classes = [FormParser,JSONParser]
queryset = Publish.objects.all()
serializer_class = PublshSerializers
def post(self, request, *args, **kwargs):
print("request.data",request.data)
return self.create(request, *args, **kwargs)
全局视图
REST_FRAMEWORK={
"DEFAULT_AUTHENTICATION_CLASSES":["app01.service.auth.Authentication",],
"DEFAULT_PERMISSION_CLASSES":["app01.service.permissions.SVIPPermission",],
"DEFAULT_THROTTLE_CLASSES":["app01.service.throttles.VisitThrottle",],
"DEFAULT_THROTTLE_RATES":{
"visit_rate":"5/m",
},
"DEFAULT_PARSER_CLASSES":[''rest_framework.parsers.FormParser'',]
}
六、url路由控制
url(r'''',include(routers.urls)),
from rest_framework import routers
routers = routers.DefaultRouter()
routers.register("authors",views.AuthorModelView)
# urls.py
from django.conf.urls import url,include
from django.contrib import admin
from app01 import views
from rest_framework import routers
routers = routers.DefaultRouter()
routers.register("authors",views.AuthorModelView)
urlpatterns = [
url(r''^admin/'', admin.site.urls),
url(r''^publishes/$'', views.PublishView.as_view(), name="publish"),
url(r''^publishes/(?P<pk>\d+)/$'', views.PublishDetailView.as_view(), name="detail_publish"),
url(r"^books/$", views.BookView.as_view(), name="books"),
url(r''^books/(?P<pk>\d+)/$'', views.BookDetailView.as_view(), name="detail_book"),
# url(r''^authors/$'',views.AuthorView.as_view(),name="author"),
# url(r''^authors/(?P<pk>\d+)'',views.AuthorDetailView.as_view(),name="detail_author"),
# url(r''^authors/$'', views.AuthorModelView.as_view({"get": "list", "post": "create"}), name="author"),
# url(r''^authors/(?P<pk>\d+)'',
# views.AuthorModelView.as_view({"get": "retrieve", "put": "update", "delete": "destroy"}), name="detail_author"),
url(r'''',include(routers.urls)),
url(r''^login/$'', views.LoginView.as_view(), name="login")
]
------------------------------------------
# views.py
from rest_framework import viewsets
class AuthorModelView(viewsets.ModelViewSet):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
效果图
七、分页
pnp = MyPageNumberPagination()
books_page = pnp.paginate_queryset(book_list,request,self)
from rest_framework.pagination import PageNumberPagination, LimitOffsetPagination
class MyPageNumberPagination(PageNumberPagination):
page_size = 2
page_query_param = "page"
page_size_query_param = "size"
max_page_size = 2 # 限制 size 得大小 不能超过多少!!
class MyLimitOffsetPagination(LimitOffsetPagination):
default_limit = 1
---------------------------
class BookView(APIView):def get(self, request):
book_list = Book.objects.all()
pnp = MyPageNumberPagination()
# pnp = MyLimitOffsetPagination()
books_page = pnp.paginate_queryset(book_list,request,self)
# ret = BookModelSerializers(book_list, many=True, context={"request": request})
ret = BookModelSerializers(books_page, many=True)
# return Response(ret.data)
return pnp.get_paginated_response(ret.data)
from rest_framework import viewsets
class AuthorModelView(viewsets.ModelViewSet):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
pagination_class = MyPageNumberPagination
效果图


day99
1 CBV
2 APIView
class BookView(APIView):pass
url(r''^books/$'', views.BookView.as_view(),name="books"),
url(r''^books/$'', View类下的view,name="books"),
一旦访问books/: view(request)======APIView类下的dispatch()====请求方式对应的示例方法()
3 def dispatch():
#一 初始化操作
# (1) 构建新的request:
self.request=self.initial_request()
# self.request._request
# self.request.GET
# self.request.data
# (2) 执行组件
# 认证,权限,频率
# 认证:request.user
self.initial(request, *args, **kwargs)
==== # 认证组件
self.perform_authentication(request)
==== request.user
=====
for authenticator in self.authenticators: # [TokenAuth(),]
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.check_permissions(request)
===========
for permission in self.get_permissions():
if not permission.has_permission(request, self):
self.permission_denied(
request, message=getattr(permission, ''message'', None)
)
# 频率组件
self.check_throttles(request)
=============
for throttle in self.get_throttles(): # [VisitRateThrottle(),]
if not throttle.allow_request(request, self):
self.throttled(request, throttle.wait()) # 受限制
# 分发
if request.method.lower() in self.http_method_names:
handler = getattr(self,request.method.lower(),
self.http_method_not_allowed)
response = handler(request, *args, **kwargs)
return response
4 序列化组件
class PublishSerializers(serializers.Serializer):
name = serializers.CharField()
email = serializers.CharField()
class PublishModelSerializers(serializers.ModelSerializer):
class Meta:
model=Publish
fields="__all__"
# queryset或者model对象-------------》json数据
ps=PublishSerializers(queryset,many=True)
ps.data # [{},{},{}]
ps=PublishSerializers(model_obj,many=False)
ps.data # {}
# json数据-------》记录
# 添加操作
ps=PublishSerializers(data=request.data)
if ps.is_valid():
ps.save() # create
# 更新操作
ps=PublishSerializers(model_obj,data=request.data)
if ps.is_valid():
ps.save() # update
5 视图组件
# 版本1:
# Book表
class BookView(APIView):
def get(self,request):
book_list=Book.objects.all()
bs=BookModelSerializers(book_list,many=True,context={''request'': request})
return Response(bs.data)
def post(self,request):
# post请求的数据
bs=BookModelSerializers(data=request.data)
if bs.is_valid():
print(bs.validated_data)
bs.save()# create方法
return Response(bs.data)
else:
return Response(bs.errors)
class BookDetailView(APIView):
def get(self,request,id):
book=Book.objects.filter(pk=id).first()
bs=BookModelSerializers(book,context={''request'': request})
return Response(bs.data)
def put(self,request,id):
book=Book.objects.filter(pk=id).first()
bs=BookModelSerializers(book,data=request.data)
if bs.is_valid():
bs.save()
return Response(bs.data)
else:
return Response(bs.errors)
def delete(self,request,id):
Book.objects.filter(pk=id).delete()
return Response()
# 版本2:mixIn
from rest_framework import mixins
from rest_framework import generics
class AuthorView(mixins.ListModelMixin,mixins.CreateModelMixin,generics.GenericAPIView):
queryset=Author.objects.all()
serializer_class =AuthorModelSerializers
def get(self,request, *args, **kwargs):
return self.list(request, *args, **kwargs)
def post(self,request, *args, **kwargs):
return self.create(request, *args, **kwargs)
class AuthorDetailView(mixins.RetrieveModelMixin,mixins.DestroyModelMixin,mixins.UpdateModelMixin,generics.GenericAPIView):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
def get(self,request,*args, **kwargs):
return self.retrieve(request,*args, **kwargs)
def delete(self,request,*args, **kwargs):
return self.destroy(request,*args, **kwargs)
def put(self,request,*args, **kwargs):
return self.retrieve(request,*args, **kwargs)
# 版本3:基于通用类
from rest_framework import mixins
from rest_framework import generics
class AuthorView(generics.ListCreateAPIView):
queryset=Author.objects.all()
serializer_class =AuthorModelSerializers
class AuthorDetailView(generics.RetrieveUpdateDestroyAPIView):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
# 版本4
class AuthorModelView(viewsets.ModelViewSet):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
url(r''^authors/$'', views.AuthorModelView.as_view({"get":"list","post":"create"}),name="author"),
url(r''^authors/(?P<pk>\d+)/$'', views.AuthorModelView.as_view({"get":"retrieve","put"
流程:
url(r''^authors/$'', views.AuthorModelView.as_view({"get":"list","post":"create"}),name="author"),
url(r''^authors/$'', ViewSetMixin.as_view({"get":"list","post":"create"}),name="author"),
url(r''^authors/$'', ViewSetMixin类下的view),
一旦访问 /authors/:
ViewSetMixin
def view():
for method, action in actions.items(): # {"get":"list","post":"create"}
handler = getattr(self, action) # self.list self.create
setattr(self, method, handler)
self.dispatch(request, *args, **kwargs)
APIView类下的self.dispatch
# 分发
if request.method.lower() in self.http_method_names:
handler = getattr(self,request.method.lower(),
self.http_method_not_allowed)
response = handler(request, *args, **kwargs) # self.list()
return response
6 认证权限频率 组件
request.META:
{''ALLUSERSPROFILE'': ''C:\\ProgramData'',
''APPDATA'': ''C:\\Users\\Administrator\\AppData\\Roaming'',
''COMMONPROGRAMFILES'': ''C:\\Program Files\\Common Files'',
''COMMONPROGRAMFILES(X86)'': ''C:\\Program Files (x86)\\Common Files'',
''COMMONPROGRAMW6432'': ''C:\\Program Files\\Common Files'',
''COMPUTERNAME'': ''PC201712041709'',
''COMSPEC'': ''C:\\Windows\\system32\\cmd.exe'',
''DJANGO_SETTINGS_MODULE'': ''restdemo.settings'',
''FP_NO_HOST_CHECK'': ''NO'', ''HOMEDRIVE'': ''C:'',
''HOMEPATH'': ''\\Users\\Administrator'',
''LOCALAPPDATA'': ''C:\\Users\\Administrator\\AppData\\Local'',
''LOGONSERVER'': ''\\\\PC201712041709'',
''NUMBER_OF_PROCESSORS'': ''4'', ''OS'': ''Windows_NT'',
''PATH'': ''C:\\Windows\\system32;C:\\Windows;C:\\Windows\\System32\\Wbem;C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\;C:\\Users\\Administrator\\AppData\\Local\\Programs\\Python\\Python36;C:\\Users\\Administrator\\AppData\\Local\\Programs\\Python\\Python36\\Scripts;C:\\Python27;E:\\MySQL Server 5.6\\bin;C:\\Users\\Administrator\\AppData\\Local\\Programs\\Python\\Python36\\Scripts\\;C:\\Users\\Administrator\\AppData\\Local\\Programs\\Python\\Python36\\;C:\\Users\\Administrator\\AppData\\Local\\atom\\bin'',
''PATHEXT'': ''.COM;.EXE;.BAT;.CMD;.VBS;.VBE;.JS;.JSE;.WSF;.WSH;.MSC'',
''PROCESSOR_ARCHITECTURE'': ''AMD64'',
''PROCESSOR_IDENTIFIER'': ''Intel64 Family 6 Model 60 Stepping 3, GenuineIntel'',
''PROCESSOR_LEVEL'': ''6'', ''PROCESSOR_REVISION'': ''3c03'',
''PROGRAMDATA'': ''C:\\ProgramData'',
''PROGRAMFILES'': ''C:\\Program Files'',
''PROGRAMFILES(X86)'': ''C:\\Program Files (x86)'',
''PROGRAMW6432'': ''C:\\Program Files'',
''PSMODULEPATH'': ''C:\\Windows\\system32\\WindowsPowerShell\\v1.0\\Modules\\'',
''PUBLIC'': ''C:\\Users\\Public'', ''PYCHARM_HOSTED'': ''1'', ''PYTHONIOENCODING'': ''UTF-8'',
''PYTHONPATH'': ''C:\\Users\\Administrator\\PycharmProjects\\s9\\restdemo'', ''PYTHONUNBUFFERED'': ''1'',
''SESSIONNAME'': ''Console'', ''SYSTEMDRIVE'': ''C:'', ''SYSTEMROOT'': ''C:\\Windows'',
''TEMP'': ''C:\\Users\\ADMINI~1\\AppData\\Local\\Temp'', ''TMP'': ''C:\\Users\\ADMINI~1\\AppData\\Local\\Temp'',
''USERDOMAIN'': ''PC201712041709'',
''USERNAME'': ''Administrator'',
''USERPROFILE'': ''C:\\Users\\Administrator'',
''WINDIR'': ''C:\\Windows'', ''WINDOWS_TRACING_FLAGS'': ''3'',
''WINDOWS_TRACING_LOGFILE'': ''C:\\BVTBin\\Tests\\installpackage\\csilogfile.log'',
''RUN_MAIN'': ''true'', ''SERVER_NAME'': ''PC201712041709'',
''GATEWAY_INTERFACE'': ''CGI/1.1'', ''SERVER_PORT'': ''8000'',
''REMOTE_HOST'': '''',
''CONTENT_LENGTH'': '''',
''SCRIPT_NAME'': '''',
''SERVER_PROTOCOL'': ''HTTP/1.1'',
''SERVER_SOFTWARE'': ''WSGIServer/0.2'',
''REQUEST_METHOD'': ''GET'',
''PATH_INFO'': ''/authors/'',
''QUERY_STRING'': ''token=8204b8e3ac40bf59ae480d17c146b51a'',
''REMOTE_ADDR'': ''127.0.0.1'',
''CONTENT_TYPE'': ''text/plain'',
''HTTP_HOST'': ''127.0.0.1:8000'',
''HTTP_CONNECTION'': ''keep-alive'',
''HTTP_USER_AGENT'': ''Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36'',
''HTTP_UPGRADE_INSECURE_REQUESTS'': ''1'',
''HTTP_ACCEPT'': ''text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8'',
''HTTP_ACCEPT_ENCODING'': ''gzip, deflate, br'', ''HTTP_ACCEPT_LANGUAGE'': ''zh-CN,zh;q=0.9'', ''HTTP_COOKIE'': ''csrftoken=jtus3l4GJEc9TFXWYCWxkBIZprcOv7C1vFMIyOHs7Zkxt015FwVZ2KEEeDV6LOyN'', ''wsgi.input'': <_io.BufferedReader name=832>, ''wsgi.errors'': <_io.TextIOWrapper name=''<stderr>'' mode=''w'' encoding=''UTF-8''>, ''wsgi.version'': (1, 0), ''wsgi.run_once'': False, ''wsgi.url_scheme'': ''http'', ''wsgi.multithread'': True, ''wsgi.multiprocess'': False, ''wsgi.file_wrapper'': <class ''wsgiref.util.FileWrapper''>}
7 解析器-----数据解析器
from rest_framework.parsers import JSONParser,FormParser,MultiPartParser,FileUploadParser
parser_classes = [JSONParser,FormParser]
8 路由控制
针对:
url(r''^authors/$'', views.AuthorModelView.as_view({"get":"list","post":"create"}),name="author"),
url(r''^authors/(?P<pk>\d+)/$'', views.AuthorModelView.as_view({"get":"retrieve","put":"update","delete":"destroy"}),name="detailauthor"),
class AuthorModelView(viewsets.ModelViewSet):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
url(r''^books/$'', views.BookModelView.as_view({"get":"list","post":"create"}),name="author"),
url(r''^books/(?P<pk>\d+)/$'', views.BookModelView.as_view({"get":"retrieve","put":"update","delete":"destroy"}),name="detailbook"),
class AuthorModelView(viewsets.ModelViewSet):
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
9 分页
10 响应器 Response


REST_FRAMEWORK = {
# ''DEFAULT_AUTHENTICATION_CLASSES'': [''app01.utils.TokenAuth''],
# ''DEFAULT_PERMISSION_CLASSES'': [''app01.utils.SVIPPermission''],
# ''DEFAULT_THROTTLE_CLASSES'': [''app01.utils.VisitThrottle''],
# "DEFAULT_THROTTLE_RATES": {
# "visit_rate": "1/m",
# }
# "PAGE_SIZE":2
}


from django.conf.urls import url,include
from django.contrib import admin
from app01 import views
from rest_framework import routers
routers = routers.DefaultRouter()
routers.register("authors",views.AuthorModelView)
urlpatterns = [
url(r''^admin/'', admin.site.urls),
url(r''^publishes/$'', views.PublishView.as_view(), name="publish"),
url(r''^publishes/(?P<pk>\d+)/$'', views.PublishDetailView.as_view(), name="detail_publish"),
url(r"^books/$", views.BookView.as_view(), name="books"),
url(r''^books/(?P<pk>\d+)/$'', views.BookDetailView.as_view(), name="detail_book"),
# url(r''^authors/$'',views.AuthorView.as_view(),name="author"),
# url(r''^authors/(?P<pk>\d+)'',views.AuthorDetailView.as_view(),name="detail_author"),
# url(r''^authors/$'', views.AuthorModelView.as_view({"get": "list", "post": "create"}), name="author"),
# url(r''^authors/(?P<pk>\d+)'',
# views.AuthorModelView.as_view({"get": "retrieve", "put": "update", "delete": "destroy"}), name="detail_author"),
url(r'''',include(routers.urls)),
url(r''^login/$'', views.LoginView.as_view(), name="login")
]


from .models import *
from rest_framework import exceptions
from rest_framework.authentication import BaseAuthentication
class TokenAuth(BaseAuthentication):
def authenticate(self,request):
token = request.GET.get("token")
token_obj = Token.objects.filter(token=token).first()
if not token_obj:
raise exceptions.AuthenticationFailed("验证失败")
return (token_obj.user.name, token_obj)
class SVIPPermission(object):
message = "只有超级用户才能访问"
def has_permission(self,request,view):
username = request.user
user_type = User.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)<3:
# history.insert(0,ctime)
# return True
# else:
# return False
#
# def wait(self):
# import time
# ctime=time.time()
# return 60-(ctime-self.history[-1])
from rest_framework.throttling import SimpleRateThrottle
class VisitThrottle(SimpleRateThrottle):
scope="visit_rate"
def get_cache_key(self, request, view):
return self.get_ident(request)


from django.shortcuts import render, HttpResponse
from django.views import View
from .models import *
import json
from rest_framework import serializers
from rest_framework.views import APIView
from rest_framework.response import Response
# 为queryset,model 对象 做序列化得
# class PublishSerializers(serializers.Serializer):
# name = serializers.CharField()
# email = serializers.CharField()
class PublishModelSerializers(serializers.ModelSerializer):
class Meta:
model = Publish
fields = "__all__"
class BookModelSerializers(serializers.ModelSerializer):
class Meta:
model = Book
fields = "__all__"
# 显示超链接
publish = serializers.HyperlinkedIdentityField(
view_name=''detailpublish'', # 别名 含正则表达式
lookup_field= ''publish_id'',
lookup_url_kwarg=''pk''
)
# publish = serializers.CharField(source="publish.pk")
# publish = serializers.CharField()
# authors = serializers.CharField(source="authors.all")
# authors = serializers.SerializerMethodField()
# def get_authors(self,obj):
# temp = []
# for obj in obj.authors.all():
# temp.append(obj.name)
# return temp
# def create(self, validated_data):
# print(''--->'',validated_data)
# book = Book.objects.create(title=validated_data["title"],price=validated_data[''price''],
# pub_date=validated_data[''pub_date''],publish_id=validated_data[''publish''][''pk''])
# book.authors.add(*validated_data[''authors''])
#
# return book
class AuthorModelSerializers(serializers.ModelSerializer):
class Meta:
model = Author
fields = "__all__"


from django.db import models
# Create your models here.
class User(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("User")
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("Publish")
authors = models.ManyToManyField("Author")
def __str__(self):
return self.title
class Publish(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


from django.shortcuts import render,HttpResponse
# Create your views here.
from django.views import View
from .models import *
import json
from rest_framework import serializers
from rest_framework.views import APIView
from rest_framework.response import Response
from app01.serializer import *
class PublishView(APIView):
def get(self, request):
publish_list = Publish.objects.all()
ret = PublishModelSerializers(publish_list, many=True)
return Response(ret.data)
def post(self, request):
ps = PublishModelSerializers(data=request.data)
if ps.is_valid():
ps.save()
return Response(ps.data)
else:
return Response(ps.errors)
class PublishDetailView(APIView):
def get(self, request, pk):
publish = Publish.objects.filter(pk=pk).first()
ps = PublishModelSerializers(publish)
return Response(ps.data)
def put(self, request, pk):
publish = Publish.objects.filter(pk=pk).first()
ps = PublishModelSerializers(publish, data=request.data)
if ps.is_valid():
ps.save()
return Response(ps.data)
else:
return Response(ps.errors)
def delete(self, request, pk):
Publish.objects.filter(pk=pk).delete()
return Response()
from rest_framework.parsers import JSONParser,FormParser,MultiPartParser,FileUploadParser
"""
默认得是 JSONParser FormParser MultiPartParser
"""
# Book
from rest_framework.pagination import PageNumberPagination,LimitOffsetPagination
class MyPageNumberPagination(PageNumberPagination):
page_size = 1
page_query_param = "page"
page_size_query_param = "size"
max_page_size = 2 # 限制 size 得大小 不能超过多少!!
# # http://127.0.0.1:8000/books/?page=2&size=2
class MyLimitOffsetPagination(LimitOffsetPagination):
default_limit = 1
# limit_query_param =
# http://127.0.0.1:8000/books/?limit=2&offset=2
# 偏移
class BookView(APIView):
# authentication_classes = [TokenAuth]
parser_classes = [JSONParser,FormParser]
def get(self,request):
print("user:--->", request.user)
print(request.auth)
book_list = Book.objects.all()
# 分页 page_size 配置 setting 全局得 单独怎么设置? 写个类
# from rest_framework.pagination import PageNumberPagination
# pnp = PageNumberPagination()
# pnp = MyPageNumberPagination()
pnp = MyLimitOffsetPagination()
books_page = pnp.paginate_queryset(book_list,request,self)
# bs = BookModelSerializers(book_list,many=True,context={''request'': request})
bs = BookModelSerializers(books_page,many=True,context={''request'': request})
# return HttpResponse(bs.data)
return Response(bs.data)
def post(self,request):
print(''data:--->'', request.data,type(request.data))
# post 请求的数据
bs = BookModelSerializers(data=request.data,context={''request'': request})
if bs.is_valid():
bs.save() # create 方法
print(bs.validated_data)
return Response(bs.data)
else:
return Response(bs.errors)
class BookDetailView(APIView):
def get(self,request,id):
book = Book.objects.filter(pk=id).first()
# 序列化
bs = BookModelSerializers(book)
return Response(bs.data)
def put(self,request,id):
book = Book.objects.filter(pk=id).first()
bs = BookModelSerializers(book,data=request.data)
if bs.is_valid():
bs.save()
return Response(bs.data)
else:
return Response(bs.errors)
def delete(self,request,id):
Book.objects.filter(pk=id).delete()
return Response()
# Author
# 逻辑复用
# 三种方法
######################### mixin类编写视图 ##############################
# from rest_framework import mixins, generics
#
# class AuthorView(mixins.ListModelMixin,mixins.CreateModelMixin,generics.GenericAPIView):
# queryset = Author.objects.all()
# serializer_class = AuthorModelSerializers
#
# def get(self,request,*args,**kwargs):
# return self.list(request,*args,**kwargs)
#
# def post(self,request,*args,**kwargs):
# return self.create(self, request, *args, **kwargs)
#
#
# class AuthorDetailView(mixins.RetrieveModelMixin,mixins.DestroyModelMixin,mixins.UpdateModelMixin,generics.GenericAPIView):
# queryset = Author.objects.all()
# serializer_class = AuthorModelSerializers
#
# def get(self,request,pk,*args,**kwargs):
# return self.retrieve(request,pk,*args,**kwargs)
#
# def delete(self,request,*args,**kwargs):
# return self.destroy(request,*args,**kwargs)
#
# def put(self,request,*args,**kwargs):
# return self.update(request,*args,**kwargs)
######################### 使用得通用得基于类得视图 ##############################
# from rest_framework import mixins, generics
#
# class AuthorView(generics.ListCreateAPIView):
# queryset = Author.objects.all()
# serializer_class = AuthorModelSerializers
#
#
# class AuthorDetailView(generics.RetrieveUpdateDestroyAPIView):
# queryset = Author.objects.all()
# serializer_class = AuthorModelSerializers
######################### viewsets.ModelViewSet ##############################
# url 需要是一趟线 走一个视图类
# url 中 利用参数 来指定 什么方式 用什么方法 执行
from .utils import TokenAuth,SVIPPermission
from rest_framework import viewsets
class VisitRateThrottle(object):
def allow_request(self,request,view):
# 要求访问站点得频率不能超过每分钟20次
if 1:
# 每次来 存下来 比对一下,间隔多久 超过一分钟
# IP, 请求首行(request.method request.path)
# 请求头(request.meta) 请求体(request.body)
print("meta:----->",request.META)
print(request.META.get("REMOTE_ADDR")) # 客户端得ip 这里面 你要保存什么 ip 时间 记录下来;
# 频率 限制 实现 功能!!!
return True
else:
return False
from rest_framework.response import Response
class AuthorModelView(viewsets.ModelViewSet):
# authentication_classes = [TokenAuth,]
# permission_classes = [SVIPPermission,] 取配全局八
# throttle_classes = [VisitRateThrottle]
queryset = Author.objects.all()
serializer_class = AuthorModelSerializers
pagination_class = MyPageNumberPagination
# 完美! ok
# 类得继承 , 表示形式 需求 展示 数据 ,覆盖方法 单独写 类得继承 可重写 覆盖
#
# def list(self,request,*args,**kwargs):
# pass
# ------------------------------------------------------
def get_random_str(user):
import hashlib,time
ctime=str(time.time())
md5=hashlib.md5(bytes(user,encoding="utf8"))
md5.update(bytes(ctime,encoding="utf8"))
return md5.hexdigest()
class LoginView(APIView):
authentication_classes = []
def post(self,request):
name = request.data.get("name")
pwd = request.data.get("pwd")
user = User.objects.filter(name=name,pwd=pwd).first()
res = {"state_code":1000,"msg":None}
if user:
random_str = get_random_str(user.name)
token = Token.objects.update_or_create(user=user,defaults={"token":random_str})
res[''token''] = random_str
else:
res["status_code"] = 1001 # 错误状态码
res[''msg''] = "用户名或密码错误"
import json
return Response(json.dumps(res,ensure_ascii=False))
八、渲染器、版本
配置:
1.添加配置
REST_FRAMEWORK = {
''DEFAULT_RENDERER_CLASSES'':[''rest_framework.renderers.JSONRenderer'',''rest_framework.renderers.BrowsableAPIRenderer''],
''DEFAULT_VERSIONING_CLASS'': ''rest_framework.versioning.URLPathVersioning'',
''ALLOWED_VERSIONS'': [''v1'', ''v2''], # 允许的版本
''VERSION_PARAM'': ''version'', # 参数
''DEFAULT_VERSION'': ''v1'', # 默认版本
}
2.设置路由:
luffycity/urls.py
from django.conf.urls import url,include
from django.contrib import admin
urlpatterns = [
# url(r''^admin/'', admin.site.urls),
url(r''^api/(?P<version>\w+)/'', include(''api.urls'')),
]
api/urls.py
from django.conf.urls import url
from api.views import course
urlpatterns = [
url(r''^course/$'', course.CourseView.as_view()),
]
3.获取版本
request.version 获取版本
Django Rest Framework
一. 什么是RESTful
- REST与技术无关,代表的是一种软件架构风格,REST是Representational State Transfer的简称,中文翻译为“表征状态转移”
- REST从资源的角度类审视整个网络,它将分布在网络中某个节点的资源通过URL进行标识,客户端应用通过URL来获取资源的表征,获得这些表征致使这些应用转变状态
- REST与技术无关,代表的是一种软件架构风格,REST是Representational State Transfer的简称,中文翻译为“表征状态转移”
- 所有的数据,不过是通过网络获取的还是操作(增删改查)的数据,都是资源,将一切数据视为资源是REST区别与其他架构风格的最本质属性
- 对于REST这种面向资源的架构风格,有人提出一种全新的结构理念,即:面向资源架构(ROA:Resource Oriented Architecture)
二. RESTful API设计
- API与用户的通信协议,总是使用HTTPs协议。
- 域名
- https://api.example.com 尽量将API部署在专用域名(会存在跨域问题)
- https://example.org/api/ API很简单
- 版本
- URL,如:https://api.example.com/v1/
- 请求头 跨域时,引发发送多次请求
- 路径,视网络上任何东西都是资源,均使用名词表示(可复数)
- https://api.example.com/v1/zoos
- https://api.example.com/v1/animals
- https://api.example.com/v1/employees
- method
- GET :从服务器取出资源(一项或多项)
- POST :在服务器新建一个资源
- PUT :在服务器更新资源(客户端提供改变后的完整资源)
- PATCH :在服务器更新资源(客户端提供改变的属性)
- DELETE :从服务器删除资源
- 过滤,通过在url上传参的形式传递搜索条件
- https://api.example.com/v1/zoos?limit=10:指定返回记录的数量
- https://api.example.com/v1/zoos?offset=10:指定返回记录的开始位置
- https://api.example.com/v1/zoos?page=2&per_page=100:指定第几页,以及每页的记录数
- https://api.example.com/v1/zoos?sortby=name&order=asc:指定返回结果按照哪个属性排序,以及排序顺序
- https://api.example.com/v1/zoos?animal_type_id=1:指定筛选条件
- 状态码
200 OK - [GET]:服务器成功返回用户请求的数据,该操作是幂等的(Idempotent)。 201 CREATED - [POST/PUT/PATCH]:用户新建或修改数据成功。 202 Accepted - [*]:表示一个请求已经进入后台排队(异步任务) 204 NO CONTENT - [DELETE]:用户删除数据成功。 400 INVALID REQUEST - [POST/PUT/PATCH]:用户发出的请求有错误,服务器没有进行新建或修改数据的操作,该操作是幂等的。 401 Unauthorized - [*]:表示用户没有权限(令牌、用户名、密码错误)。 403 Forbidden - [*] 表示用户得到授权(与401错误相对),但是访问是被禁止的。 404 NOT FOUND - [*]:用户发出的请求针对的是不存在的记录,服务器没有进行操作,该操作是幂等的。 406 Not Acceptable - [GET]:用户请求的格式不可得(比如用户请求JSON格式,但是只有XML格式)。 410 Gone -[GET]:用户请求的资源被永久删除,且不会再得到的。 422 Unprocesable entity - [POST/PUT/PATCH] 当创建一个对象时,发生一个验证错误。 500 INTERNAL SERVER ERROR - [*]:服务器发生错误,用户将无法判断发出的请求是否成功。 更多看这里:http://www.w3.org/Protocols/rfc2616/rfc2616-sec10.html
- 错误处理,状态码是4xx时,应返回错误信息,error当做key。
-
{
error:
"Invalid API key"
}
-
- 返回结果,针对不同操作,服务器向用户返回的结果应该符合以下规范。
-
GET
/
collection:返回资源对象的列表(数组)
GET
/
collection
/
resource:返回单个资源对象
POST
/
collection:返回新生成的资源对象
PUT
/
collection
/
resource:返回完整的资源对象
PATCH
/
collection
/
resource:返回完整的资源对象
DELETE
/
collection
/
resource:返回一个空文档
- Hypermedia API,RESTful API最好做到Hypermedia,即返回结果中提供链接,连向其他API方法,使得用户不查文档,也知道下一步应该做什么。
-
{
"link"
: {
"rel"
:
"collection https://www.example.com/zoos"
,
"href"
:
"https://api.example.com/zoos"
,
"title"
:
"List of zoos"
,
"type"
:
"application/vnd.yourformat+json"
}}
三. 基于Django实现
路由系统:
urlpatterns = [
url(r''^users'', Users.as_view()),
]
CBV视图:
from django.views import View
from django.http import JsonResponse
class Users(View):
def get(self, request, *args, **kwargs):
result = {
''status'': True,
''data'': ''response data''
}
return JsonResponse(result, status=200)
def post(self, request, *args, **kwargs):
result = {
''status'': True,
''data'': ''response data''
}
return JsonResponse(result, status=200)
四. 基于Django Rest Framework框架实现
1. 基本流程
url.py
from django.conf.urls import url, include
from web.views.s1_api 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等方法
注意:APIView中的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请求,响应内容'')
上述是rest framework框架基本流程,重要的功能是在APIView的dispatch中触发。
2. 认证和授权
a. 用户url传入的token认证


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


from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.authentication import BaseAuthentication
from rest_framework.request import Request
from rest_framework import exceptions
token_list = [
''sfsfss123kuf3j123'',
''asijnfowerkkf9812'',
]
class TestAuthentication(BaseAuthentication):
def authenticate(self, request):
"""
用户认证,如果验证成功后返回元组: (用户,用户Token)
:param request:
:return:
None,表示跳过该验证;
如果跳过了所有认证,默认用户和Token和使用配置文件进行设置
self._authenticator = None
if api_settings.UNAUTHENTICATED_USER:
self.user = api_settings.UNAUTHENTICATED_USER()
else:
self.user = None
if api_settings.UNAUTHENTICATED_TOKEN:
self.auth = api_settings.UNAUTHENTICATED_TOKEN()
else:
self.auth = None
(user,token)表示验证通过并设置用户名和Token;
AuthenticationFailed异常
"""
val = request.query_params.get(''token'')
if val not in token_list:
raise exceptions.AuthenticationFailed("用户认证失败")
return (''登录用户'', ''用户token'')
def authenticate_header(self, request):
"""
Return a string to be used as the value of the `WWW-Authenticate`
header in a `401 Unauthenticated` response, or `None` if the
authentication scheme should return `403 Permission Denied` responses.
"""
# 验证失败时,返回的响应头WWW-Authenticate对应的值
pass
class TestView(APIView):
authentication_classes = [TestAuthentication, ]
permission_classes = []
def get(self, request, *args, **kwargs):
print(request.user)
print(request.auth)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
b. 请求头认证


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.authentication import BaseAuthentication
from rest_framework.request import Request
from rest_framework import exceptions
token_list = [
''sfsfss123kuf3j123'',
''asijnfowerkkf9812'',
]
class TestAuthentication(BaseAuthentication):
def authenticate(self, request):
"""
用户认证,如果验证成功后返回元组: (用户,用户Token)
:param request:
:return:
None,表示跳过该验证;
如果跳过了所有认证,默认用户和Token和使用配置文件进行设置
self._authenticator = None
if api_settings.UNAUTHENTICATED_USER:
self.user = api_settings.UNAUTHENTICATED_USER()
else:
self.user = None
if api_settings.UNAUTHENTICATED_TOKEN:
self.auth = api_settings.UNAUTHENTICATED_TOKEN()
else:
self.auth = None
(user,token)表示验证通过并设置用户名和Token;
AuthenticationFailed异常
"""
import base64
auth = request.META.get(''HTTP_AUTHORIZATION'', b'''')
if auth:
auth = auth.encode(''utf-8'')
auth = auth.split()
if not auth or auth[0].lower() != b''basic'':
raise exceptions.AuthenticationFailed(''验证失败'')
if len(auth) != 2:
raise exceptions.AuthenticationFailed(''验证失败'')
username, part, password = base64.b64decode(auth[1]).decode(''utf-8'').partition('':'')
if username == ''alex'' and password == ''123'':
return (''登录用户'', ''用户token'')
else:
raise exceptions.AuthenticationFailed(''用户名或密码错误'')
def authenticate_header(self, request):
"""
Return a string to be used as the value of the `WWW-Authenticate`
header in a `401 Unauthenticated` response, or `None` if the
authentication scheme should return `403 Permission Denied` responses.
"""
return ''Basic realm=api''
class TestView(APIView):
authentication_classes = [TestAuthentication, ]
permission_classes = []
def get(self, request, *args, **kwargs):
print(request.user)
print(request.auth)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
c. 多个认证规则


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.authentication import BaseAuthentication
from rest_framework.request import Request
from rest_framework import exceptions
token_list = [
''sfsfss123kuf3j123'',
''asijnfowerkkf9812'',
]
class Test1Authentication(BaseAuthentication):
def authenticate(self, request):
"""
用户认证,如果验证成功后返回元组: (用户,用户Token)
:param request:
:return:
None,表示跳过该验证;
如果跳过了所有认证,默认用户和Token和使用配置文件进行设置
self._authenticator = None
if api_settings.UNAUTHENTICATED_USER:
self.user = api_settings.UNAUTHENTICATED_USER() # 默认值为:匿名用户
else:
self.user = None
if api_settings.UNAUTHENTICATED_TOKEN:
self.auth = api_settings.UNAUTHENTICATED_TOKEN()# 默认值为:None
else:
self.auth = None
(user,token)表示验证通过并设置用户名和Token;
AuthenticationFailed异常
"""
import base64
auth = request.META.get(''HTTP_AUTHORIZATION'', b'''')
if auth:
auth = auth.encode(''utf-8'')
else:
return None
print(auth,''xxxx'')
auth = auth.split()
if not auth or auth[0].lower() != b''basic'':
raise exceptions.AuthenticationFailed(''验证失败'')
if len(auth) != 2:
raise exceptions.AuthenticationFailed(''验证失败'')
username, part, password = base64.b64decode(auth[1]).decode(''utf-8'').partition('':'')
if username == ''alex'' and password == ''123'':
return (''登录用户'', ''用户token'')
else:
raise exceptions.AuthenticationFailed(''用户名或密码错误'')
def authenticate_header(self, request):
"""
Return a string to be used as the value of the `WWW-Authenticate`
header in a `401 Unauthenticated` response, or `None` if the
authentication scheme should return `403 Permission Denied` responses.
"""
# return ''Basic realm=api''
pass
class Test2Authentication(BaseAuthentication):
def authenticate(self, request):
"""
用户认证,如果验证成功后返回元组: (用户,用户Token)
:param request:
:return:
None,表示跳过该验证;
如果跳过了所有认证,默认用户和Token和使用配置文件进行设置
self._authenticator = None
if api_settings.UNAUTHENTICATED_USER:
self.user = api_settings.UNAUTHENTICATED_USER() # 默认值为:匿名用户
else:
self.user = None
if api_settings.UNAUTHENTICATED_TOKEN:
self.auth = api_settings.UNAUTHENTICATED_TOKEN()# 默认值为:None
else:
self.auth = None
(user,token)表示验证通过并设置用户名和Token;
AuthenticationFailed异常
"""
val = request.query_params.get(''token'')
if val not in token_list:
raise exceptions.AuthenticationFailed("用户认证失败")
return (''登录用户'', ''用户token'')
def authenticate_header(self, request):
"""
Return a string to be used as the value of the `WWW-Authenticate`
header in a `401 Unauthenticated` response, or `None` if the
authentication scheme should return `403 Permission Denied` responses.
"""
pass
class TestView(APIView):
authentication_classes = [Test1Authentication, Test2Authentication]
permission_classes = []
def get(self, request, *args, **kwargs):
print(request.user)
print(request.auth)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
d. 认证和权限


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.authentication import BaseAuthentication
from rest_framework.permissions import BasePermission
from rest_framework.request import Request
from rest_framework import exceptions
token_list = [
''sfsfss123kuf3j123'',
''asijnfowerkkf9812'',
]
class TestAuthentication(BaseAuthentication):
def authenticate(self, request):
"""
用户认证,如果验证成功后返回元组: (用户,用户Token)
:param request:
:return:
None,表示跳过该验证;
如果跳过了所有认证,默认用户和Token和使用配置文件进行设置
self._authenticator = None
if api_settings.UNAUTHENTICATED_USER:
self.user = api_settings.UNAUTHENTICATED_USER() # 默认值为:匿名用户
else:
self.user = None
if api_settings.UNAUTHENTICATED_TOKEN:
self.auth = api_settings.UNAUTHENTICATED_TOKEN()# 默认值为:None
else:
self.auth = None
(user,token)表示验证通过并设置用户名和Token;
AuthenticationFailed异常
"""
val = request.query_params.get(''token'')
if val not in token_list:
raise exceptions.AuthenticationFailed("用户认证失败")
return (''登录用户'', ''用户token'')
def authenticate_header(self, request):
"""
Return a string to be used as the value of the `WWW-Authenticate`
header in a `401 Unauthenticated` response, or `None` if the
authentication scheme should return `403 Permission Denied` responses.
"""
pass
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触发
authentication_classes = [TestAuthentication, ]
# 权限
# 循环执行所有的权限
permission_classes = [TestPermission, ]
def get(self, request, *args, **kwargs):
# self.dispatch
print(request.user)
print(request.auth)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
e. 全局使用
上述操作中均是对单独视图进行特殊配置,如果想要对全局进行配置,则需要再配置文件中写入即可。


REST_FRAMEWORK = {
''UNAUTHENTICATED_USER'': None,
''UNAUTHENTICATED_TOKEN'': None,
"DEFAULT_AUTHENTICATION_CLASSES": [
"web.utils.TestAuthentication",
],
"DEFAULT_PERMISSION_CLASSES": [
"web.utils.TestPermission",
],
}


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
class TestView(APIView):
def get(self, request, *args, **kwargs):
# self.dispatch
print(request.user)
print(request.auth)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
3. 用户访问次数/频率限制
a. 基于用户IP限制访问频率


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import exceptions
from rest_framework.throttling import BaseThrottle
from rest_framework.settings import api_settings
# 保存访问记录
RECORD = {
''用户IP'': [12312139, 12312135, 12312133, ]
}
class TestThrottle(BaseThrottle):
ctime = time.time
def get_ident(self, request):
"""
根据用户IP和代理IP,当做请求者的唯一IP
Identify the machine making the request by parsing HTTP_X_FORWARDED_FOR
if present and number of proxies is > 0. If not use all of
HTTP_X_FORWARDED_FOR if it is available, if not use REMOTE_ADDR.
"""
xff = request.META.get(''HTTP_X_FORWARDED_FOR'')
remote_addr = request.META.get(''REMOTE_ADDR'')
num_proxies = api_settings.NUM_PROXIES
if num_proxies is not None:
if num_proxies == 0 or xff is None:
return remote_addr
addrs = xff.split('','')
client_addr = addrs[-min(num_proxies, len(addrs))]
return client_addr.strip()
return ''''.join(xff.split()) if xff else remote_addr
def allow_request(self, request, view):
"""
是否仍然在允许范围内
Return `True` if the request should be allowed, `False` otherwise.
:param request:
:param view:
:return: True,表示可以通过;False表示已超过限制,不允许访问
"""
# 获取用户唯一标识(如:IP)
# 允许一分钟访问10次
num_request = 10
time_request = 60
now = self.ctime()
ident = self.get_ident(request)
self.ident = ident
if ident not in RECORD:
RECORD[ident] = [now, ]
return True
history = RECORD[ident]
while history and history[-1] <= now - time_request:
history.pop()
if len(history) < num_request:
history.insert(0, now)
return True
def wait(self):
"""
多少秒后可以允许继续访问
Optionally, return a recommended number of seconds to wait before
the next request.
"""
last_time = RECORD[self.ident][0]
now = self.ctime()
return int(60 + last_time - now)
class TestView(APIView):
throttle_classes = [TestThrottle, ]
def get(self, request, *args, **kwargs):
# self.dispatch
print(request.user)
print(request.auth)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
def throttled(self, request, wait):
"""
访问次数被限制时,定制错误信息
"""
class Throttled(exceptions.Throttled):
default_detail = ''请求被限制.''
extra_detail_singular = ''请 {wait} 秒之后再重试.''
extra_detail_plural = ''请 {wait} 秒之后再重试.''
raise Throttled(wait)
b. 基于用户IP显示访问频率(利于Django缓存)


REST_FRAMEWORK = {
''DEFAULT_THROTTLE_RATES'': {
''test_scope'': ''10/m'',
},
}


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import exceptions
from rest_framework.throttling import SimpleRateThrottle
class TestThrottle(SimpleRateThrottle):
# 配置文件定义的显示频率的Key
scope = "test_scope"
def get_cache_key(self, request, view):
"""
Should return a unique cache-key which can be used for throttling.
Must be overridden.
May return `None` if the request should not be throttled.
"""
if not request.user:
ident = self.get_ident(request)
else:
ident = request.user
return self.cache_format % {
''scope'': self.scope,
''ident'': ident
}
class TestView(APIView):
throttle_classes = [TestThrottle, ]
def get(self, request, *args, **kwargs):
# self.dispatch
print(request.user)
print(request.auth)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
def throttled(self, request, wait):
"""
访问次数被限制时,定制错误信息
"""
class Throttled(exceptions.Throttled):
default_detail = ''请求被限制.''
extra_detail_singular = ''请 {wait} 秒之后再重试.''
extra_detail_plural = ''请 {wait} 秒之后再重试.''
raise Throttled(wait)
c. view中限制请求频率


REST_FRAMEWORK = {
''DEFAULT_THROTTLE_RATES'': {
''xxxxxx'': ''10/m'',
},
}


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import exceptions
from rest_framework.throttling import ScopedRateThrottle
# 继承 ScopedRateThrottle
class TestThrottle(ScopedRateThrottle):
def get_cache_key(self, request, view):
"""
Should return a unique cache-key which can be used for throttling.
Must be overridden.
May return `None` if the request should not be throttled.
"""
if not request.user:
ident = self.get_ident(request)
else:
ident = request.user
return self.cache_format % {
''scope'': self.scope,
''ident'': ident
}
class TestView(APIView):
throttle_classes = [TestThrottle, ]
# 在settings中获取 xxxxxx 对应的频率限制值
throttle_scope = "xxxxxx"
def get(self, request, *args, **kwargs):
# self.dispatch
print(request.user)
print(request.auth)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
def throttled(self, request, wait):
"""
访问次数被限制时,定制错误信息
"""
class Throttled(exceptions.Throttled):
default_detail = ''请求被限制.''
extra_detail_singular = ''请 {wait} 秒之后再重试.''
extra_detail_plural = ''请 {wait} 秒之后再重试.''
raise Throttled(wait)
d. 匿名时用IP限制+登录时用Token限制


REST_FRAMEWORK = {
''UNAUTHENTICATED_USER'': None,
''UNAUTHENTICATED_TOKEN'': None,
''DEFAULT_THROTTLE_RATES'': {
''luffy_anon'': ''10/m'',
''luffy_user'': ''20/m'',
},
}


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.throttling import SimpleRateThrottle
class LuffyAnonRateThrottle(SimpleRateThrottle):
"""
匿名用户,根据IP进行限制
"""
scope = "luffy_anon"
def get_cache_key(self, request, view):
# 用户已登录,则跳过 匿名频率限制
if request.user:
return None
return self.cache_format % {
''scope'': self.scope,
''ident'': self.get_ident(request)
}
class LuffyUserRateThrottle(SimpleRateThrottle):
"""
登录用户,根据用户token限制
"""
scope = "luffy_user"
def get_ident(self, request):
"""
认证成功时:request.user是用户对象;request.auth是token对象
:param request:
:return:
"""
# return request.auth.token
return "user_token"
def get_cache_key(self, request, view):
"""
获取缓存key
:param request:
:param view:
:return:
"""
# 未登录用户,则跳过 Token限制
if not request.user:
return None
return self.cache_format % {
''scope'': self.scope,
''ident'': self.get_ident(request)
}
class TestView(APIView):
throttle_classes = [LuffyUserRateThrottle, LuffyAnonRateThrottle, ]
def get(self, request, *args, **kwargs):
# self.dispatch
print(request.user)
print(request.auth)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
e. 全局使用


REST_FRAMEWORK = {
''DEFAULT_THROTTLE_CLASSES'': [
''api.utils.throttles.throttles.LuffyAnonRateThrottle'',
''api.utils.throttles.throttles.LuffyUserRateThrottle'',
],
''DEFAULT_THROTTLE_RATES'': {
''anon'': ''10/day'',
''user'': ''10/day'',
''luffy_anon'': ''10/m'',
''luffy_user'': ''20/m'',
},
}
4. 版本
a. 基于url的get传参方式
如:/users?version=v1


REST_FRAMEWORK = {
''DEFAULT_VERSION'': ''v1'', # 默认版本
''ALLOWED_VERSIONS'': [''v1'', ''v2''], # 允许的版本
''VERSION_PARAM'': ''version'' # URL中获取值的key
}


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.versioning import QueryParameterVersioning
class TestView(APIView):
versioning_class = QueryParameterVersioning
def get(self, request, *args, **kwargs):
# 获取版本
print(request.version)
# 获取版本管理的类
print(request.versioning_scheme)
# 反向生成URL
reverse_url = request.versioning_scheme.reverse(''test'', request=request)
print(reverse_url)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
b. 基于url的正则方式
如:/v1/users/


REST_FRAMEWORK = {
''DEFAULT_VERSION'': ''v1'', # 默认版本
''ALLOWED_VERSIONS'': [''v1'', ''v2''], # 允许的版本
''VERSION_PARAM'': ''version'' # URL中获取值的key
}


from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''^(?P<version>[v1|v2]+)/test/'', TestView.as_view(), name=''test''),
]


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.versioning import URLPathVersioning
class TestView(APIView):
versioning_class = URLPathVersioning
def get(self, request, *args, **kwargs):
# 获取版本
print(request.version)
# 获取版本管理的类
print(request.versioning_scheme)
# 反向生成URL
reverse_url = request.versioning_scheme.reverse(''test'', request=request)
print(reverse_url)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
c. 基于 accept 请求头方式
如:Accept: application/json; version=1.0


REST_FRAMEWORK = {
''DEFAULT_VERSION'': ''v1'', # 默认版本
''ALLOWED_VERSIONS'': [''v1'', ''v2''], # 允许的版本
''VERSION_PARAM'': ''version'' # URL中获取值的key
}


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.versioning import AcceptHeaderVersioning
class TestView(APIView):
versioning_class = AcceptHeaderVersioning
def get(self, request, *args, **kwargs):
# 获取版本 HTTP_ACCEPT头
print(request.version)
# 获取版本管理的类
print(request.versioning_scheme)
# 反向生成URL
reverse_url = request.versioning_scheme.reverse(''test'', request=request)
print(reverse_url)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
d. 基于主机名方法
如:v1.example.com


ALLOWED_HOSTS = [''*'']
REST_FRAMEWORK = {
''DEFAULT_VERSION'': ''v1'', # 默认版本
''ALLOWED_VERSIONS'': [''v1'', ''v2''], # 允许的版本
''VERSION_PARAM'': ''version'' # URL中获取值的key
}


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.versioning import HostNameVersioning
class TestView(APIView):
versioning_class = HostNameVersioning
def get(self, request, *args, **kwargs):
# 获取版本
print(request.version)
# 获取版本管理的类
print(request.versioning_scheme)
# 反向生成URL
reverse_url = request.versioning_scheme.reverse(''test'', request=request)
print(reverse_url)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
e. 基于django路由系统的namespace
如:example.com/v1/users/


REST_FRAMEWORK = {
''DEFAULT_VERSION'': ''v1'', # 默认版本
''ALLOWED_VERSIONS'': [''v1'', ''v2''], # 允许的版本
''VERSION_PARAM'': ''version'' # URL中获取值的key
}


from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''^v1/'', ([
url(r''test/'', TestView.as_view(), name=''test''),
], None, ''v1'')),
url(r''^v2/'', ([
url(r''test/'', TestView.as_view(), name=''test''),
], None, ''v2'')),
]


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.versioning import NamespaceVersioning
class TestView(APIView):
versioning_class = NamespaceVersioning
def get(self, request, *args, **kwargs):
# 获取版本
print(request.version)
# 获取版本管理的类
print(request.versioning_scheme)
# 反向生成URL
reverse_url = request.versioning_scheme.reverse(''test'', request=request)
print(reverse_url)
return Response(''GET请求,响应内容'')
def post(self, request, *args, **kwargs):
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
f. 全局使用


REST_FRAMEWORK = {
''DEFAULT_VERSIONING_CLASS'':"rest_framework.versioning.URLPathVersioning",
''DEFAULT_VERSION'': ''v1'',
''ALLOWED_VERSIONS'': [''v1'', ''v2''],
''VERSION_PARAM'': ''version''
}
5. 解析器(parser)
根据请求头 content-type 选择对应的解析器就请求体内容进行处理。
a. 仅处理请求头content-type为application/json的请求体


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.request import Request
from rest_framework.parsers import JSONParser
class TestView(APIView):
parser_classes = [JSONParser, ]
def post(self, request, *args, **kwargs):
print(request.content_type)
# 获取请求的值,并使用对应的JSONParser进行处理
print(request.data)
# application/x-www-form-urlencoded 或 multipart/form-data时,request.POST中才有值
print(request.POST)
print(request.FILES)
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
b. 仅处理请求头content-type为application/x-www-form-urlencoded 的请求体


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.request import Request
from rest_framework.parsers import FormParser
class TestView(APIView):
parser_classes = [FormParser, ]
def post(self, request, *args, **kwargs):
print(request.content_type)
# 获取请求的值,并使用对应的JSONParser进行处理
print(request.data)
# application/x-www-form-urlencoded 或 multipart/form-data时,request.POST中才有值
print(request.POST)
print(request.FILES)
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
c. 仅处理请求头content-type为multipart/form-data的请求体


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.request import Request
from rest_framework.parsers import MultiPartParser
class TestView(APIView):
parser_classes = [MultiPartParser, ]
def post(self, request, *args, **kwargs):
print(request.content_type)
# 获取请求的值,并使用对应的JSONParser进行处理
print(request.data)
# application/x-www-form-urlencoded 或 multipart/form-data时,request.POST中才有值
print(request.POST)
print(request.FILES)
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')


<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form action="http://127.0.0.1:8000/test/" method="post" enctype="multipart/form-data">
<input type="text" name="user" />
<input type="file" name="img">
<input type="submit" value="提交">
</form>
</body>
</html>
d. 仅上传文件


from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''test/(?P<filename>[^/]+)'', TestView.as_view(), name=''test''),
]


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.request import Request
from rest_framework.parsers import FileUploadParser
class TestView(APIView):
parser_classes = [FileUploadParser, ]
def post(self, request, filename, *args, **kwargs):
print(filename)
print(request.content_type)
# 获取请求的值,并使用对应的JSONParser进行处理
print(request.data)
# application/x-www-form-urlencoded 或 multipart/form-data时,request.POST中才有值
print(request.POST)
print(request.FILES)
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')


<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form action="http://127.0.0.1:8000/test/f1.numbers" method="post" enctype="multipart/form-data">
<input type="text" name="user" />
<input type="file" name="img">
<input type="submit" value="提交">
</form>
</body>
</html>
e. 同时多个Parser
当同时使用多个parser时,rest framework会根据请求头content-type自动进行比对,并使用对应parser


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.request import Request
from rest_framework.parsers import JSONParser, FormParser, MultiPartParser
class TestView(APIView):
parser_classes = [JSONParser, FormParser, MultiPartParser, ]
def post(self, request, *args, **kwargs):
print(request.content_type)
# 获取请求的值,并使用对应的JSONParser进行处理
print(request.data)
# application/x-www-form-urlencoded 或 multipart/form-data时,request.POST中才有值
print(request.POST)
print(request.FILES)
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
f. 全局使用


REST_FRAMEWORK = {
''DEFAULT_PARSER_CLASSES'':[
''rest_framework.parsers.JSONParser''
''rest_framework.parsers.FormParser''
''rest_framework.parsers.MultiPartParser''
]
}


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
class TestView(APIView):
def post(self, request, *args, **kwargs):
print(request.content_type)
# 获取请求的值,并使用对应的JSONParser进行处理
print(request.data)
# application/x-www-form-urlencoded 或 multipart/form-data时,request.POST中才有值
print(request.POST)
print(request.FILES)
return Response(''POST请求,响应内容'')
def put(self, request, *args, **kwargs):
return Response(''PUT请求,响应内容'')
注意:个别特殊的值可以通过Django的request对象 request._request 来进行获取
6. 序列化
序列化用于对用户请求数据进行验证和数据进行序列化。
a. 自定义字段


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import serializers
from .. import models
class PasswordValidator(object):
def __init__(self, base):
self.base = base
def __call__(self, value):
if value != self.base:
message = ''This field must be %s.'' % self.base
raise serializers.ValidationError(message)
def set_context(self, serializer_field):
"""
This hook is called by the serializer instance,
prior to the validation call being made.
"""
# 执行验证之前调用,serializer_fields是当前字段对象
pass
class UserSerializer(serializers.Serializer):
ut_title = serializers.CharField(source=''ut.title'')
user = serializers.CharField(min_length=6)
pwd = serializers.CharField(error_messages={''required'': ''密码不能为空''}, validators=[PasswordValidator(''666'')])
class TestView(APIView):
def get(self, request, *args, **kwargs):
# 序列化,将数据库查询字段序列化为字典
data_list = models.UserInfo.objects.all()
ser = UserSerializer(instance=data_list, many=True)
# 或
# obj = models.UserInfo.objects.all().first()
# ser = UserSerializer(instance=obj, many=False)
return Response(ser.data)
def post(self, request, *args, **kwargs):
# 验证,对请求发来的数据进行验证
ser = UserSerializer(data=request.data)
if ser.is_valid():
print(ser.validated_data)
else:
print(ser.errors)
return Response(''POST请求,响应内容'')
b. 基于Model自动生成字段


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import serializers
from .. import models
class PasswordValidator(object):
def __init__(self, base):
self.base = str(base)
def __call__(self, value):
if value != self.base:
message = ''This field must be %s.'' % self.base
raise serializers.ValidationError(message)
def set_context(self, serializer_field):
"""
This hook is called by the serializer instance,
prior to the validation call being made.
"""
# 执行验证之前调用,serializer_fields是当前字段对象
pass
class ModelUserSerializer(serializers.ModelSerializer):
user = serializers.CharField(max_length=32)
class Meta:
model = models.UserInfo
fields = "__all__"
# fields = [''user'', ''pwd'', ''ut'']
depth = 2
extra_kwargs = {''user'': {''min_length'': 6}, ''pwd'': {''validators'': [PasswordValidator(666), ]}}
# read_only_fields = [''user'']
class TestView(APIView):
def get(self, request, *args, **kwargs):
# 序列化,将数据库查询字段序列化为字典
data_list = models.UserInfo.objects.all()
ser = ModelUserSerializer(instance=data_list, many=True)
# 或
# obj = models.UserInfo.objects.all().first()
# ser = UserSerializer(instance=obj, many=False)
return Response(ser.data)
def post(self, request, *args, **kwargs):
# 验证,对请求发来的数据进行验证
print(request.data)
ser = ModelUserSerializer(data=request.data)
if ser.is_valid():
print(ser.validated_data)
else:
print(ser.errors)
return Response(''POST请求,响应内容'')
c. 生成URL


from django.conf.urls import url, include
from web.views.s6_serializers import TestView
urlpatterns = [
url(r''test/'', TestView.as_view(), name=''test''),
url(r''detail/(?P<pk>\d+)/'', TestView.as_view(), name=''detail''),
]


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import serializers
from .. import models
class PasswordValidator(object):
def __init__(self, base):
self.base = str(base)
def __call__(self, value):
if value != self.base:
message = ''This field must be %s.'' % self.base
raise serializers.ValidationError(message)
def set_context(self, serializer_field):
"""
This hook is called by the serializer instance,
prior to the validation call being made.
"""
# 执行验证之前调用,serializer_fields是当前字段对象
pass
class ModelUserSerializer(serializers.ModelSerializer):
ut = serializers.HyperlinkedIdentityField(view_name=''detail'')
class Meta:
model = models.UserInfo
fields = "__all__"
extra_kwargs = {
''user'': {''min_length'': 6},
''pwd'': {''validators'': [PasswordValidator(666),]},
}
class TestView(APIView):
def get(self, request, *args, **kwargs):
# 序列化,将数据库查询字段序列化为字典
data_list = models.UserInfo.objects.all()
ser = ModelUserSerializer(instance=data_list, many=True, context={''request'': request})
# 或
# obj = models.UserInfo.objects.all().first()
# ser = UserSerializer(instance=obj, many=False)
return Response(ser.data)
def post(self, request, *args, **kwargs):
# 验证,对请求发来的数据进行验证
print(request.data)
ser = ModelUserSerializer(data=request.data)
if ser.is_valid():
print(ser.validated_data)
else:
print(ser.errors)
return Response(''POST请求,响应内容'')
d. 自动生成URL


from django.conf.urls import url, include
from web.views.s6_serializers import TestView
urlpatterns = [
url(r''test/'', TestView.as_view(), name=''test''),
url(r''detail/(?P<pk>\d+)/'', TestView.as_view(), name=''xxxx''),
]


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import serializers
from .. import models
class PasswordValidator(object):
def __init__(self, base):
self.base = str(base)
def __call__(self, value):
if value != self.base:
message = ''This field must be %s.'' % self.base
raise serializers.ValidationError(message)
def set_context(self, serializer_field):
"""
This hook is called by the serializer instance,
prior to the validation call being made.
"""
# 执行验证之前调用,serializer_fields是当前字段对象
pass
class ModelUserSerializer(serializers.HyperlinkedModelSerializer):
ll = serializers.HyperlinkedIdentityField(view_name=''xxxx'')
tt = serializers.CharField(required=False)
class Meta:
model = models.UserInfo
fields = "__all__"
list_serializer_class = serializers.ListSerializer
extra_kwargs = {
''user'': {''min_length'': 6},
''pwd'': {''validators'': [PasswordValidator(666), ]},
''url'': {''view_name'': ''xxxx''},
''ut'': {''view_name'': ''xxxx''},
}
class TestView(APIView):
def get(self, request, *args, **kwargs):
# # 序列化,将数据库查询字段序列化为字典
data_list = models.UserInfo.objects.all()
ser = ModelUserSerializer(instance=data_list, many=True, context={''request'': request})
# # 如果Many=True
# # 或
# # obj = models.UserInfo.objects.all().first()
# # ser = UserSerializer(instance=obj, many=False)
return Response(ser.data)
def post(self, request, *args, **kwargs):
# 验证,对请求发来的数据进行验证
print(request.data)
ser = ModelUserSerializer(data=request.data)
if ser.is_valid():
print(ser.validated_data)
else:
print(ser.errors)
return Response(''POST请求,响应内容'')
7. 分页
a. 根据页码进行分页


from django.conf.urls import url, include
from rest_framework import routers
from web.views import s9_pagination
urlpatterns = [
url(r''^test/'', s9_pagination.UserViewSet.as_view()),
]


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework import serializers
from .. import models
from rest_framework.pagination import PageNumberPagination
class StandardResultsSetPagination(PageNumberPagination):
# 默认每页显示的数据条数
page_size = 1
# 获取URL参数中设置的每页显示数据条数
page_size_query_param = ''page_size''
# 获取URL参数中传入的页码key
page_query_param = ''page''
# 最大支持的每页显示的数据条数
max_page_size = 1
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = models.UserInfo
fields = "__all__"
class UserViewSet(APIView):
def get(self, request, *args, **kwargs):
user_list = models.UserInfo.objects.all().order_by(''-id'')
# 实例化分页对象,获取数据库中的分页数据
paginator = StandardResultsSetPagination()
page_user_list = paginator.paginate_queryset(user_list, self.request, view=self)
# 序列化对象
serializer = UserSerializer(page_user_list, many=True)
# 生成分页和数据
response = paginator.get_paginated_response(serializer.data)
return response
b. 位置和个数进行分页


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework import serializers
from .. import models
from rest_framework.pagination import PageNumberPagination,LimitOffsetPagination
class StandardResultsSetPagination(LimitOffsetPagination):
# 默认每页显示的数据条数
default_limit = 10
# URL中传入的显示数据条数的参数
limit_query_param = ''limit''
# URL中传入的数据位置的参数
offset_query_param = ''offset''
# 最大每页显得条数
max_limit = None
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = models.UserInfo
fields = "__all__"
class UserViewSet(APIView):
def get(self, request, *args, **kwargs):
user_list = models.UserInfo.objects.all().order_by(''-id'')
# 实例化分页对象,获取数据库中的分页数据
paginator = StandardResultsSetPagination()
page_user_list = paginator.paginate_queryset(user_list, self.request, view=self)
# 序列化对象
serializer = UserSerializer(page_user_list, many=True)
# 生成分页和数据
response = paginator.get_paginated_response(serializer.data)
return response
c. 游标分页


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


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework import serializers
from .. import models
from rest_framework.pagination import PageNumberPagination, LimitOffsetPagination, CursorPagination
class StandardResultsSetPagination(CursorPagination):
# URL传入的游标参数
cursor_query_param = ''cursor''
# 默认每页显示的数据条数
page_size = 2
# URL传入的每页显示条数的参数
page_size_query_param = ''page_size''
# 每页显示数据最大条数
max_page_size = 1000
# 根据ID从大到小排列
ordering = "id"
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = models.UserInfo
fields = "__all__"
class UserViewSet(APIView):
def get(self, request, *args, **kwargs):
user_list = models.UserInfo.objects.all().order_by(''-id'')
# 实例化分页对象,获取数据库中的分页数据
paginator = StandardResultsSetPagination()
page_user_list = paginator.paginate_queryset(user_list, self.request, view=self)
# 序列化对象
serializer = UserSerializer(page_user_list, many=True)
# 生成分页和数据
response = paginator.get_paginated_response(serializer.data)
return response
8. 路由系统
a. 自定义路由


from django.conf.urls import url, include
from web.views import s11_render
urlpatterns = [
url(r''^test/$'', s11_render.TestView.as_view()),
url(r''^test\.(?P<format>[a-z0-9]+)$'', s11_render.TestView.as_view()),
url(r''^test/(?P<pk>[^/.]+)/$'', s11_render.TestView.as_view()),
url(r''^test/(?P<pk>[^/.]+)\.(?P<format>[a-z0-9]+)$'', s11_render.TestView.as_view())
]


from rest_framework.views import APIView
from rest_framework.response import Response
from .. import models
class TestView(APIView):
def get(self, request, *args, **kwargs):
print(kwargs)
print(self.renderer_classes)
return Response(''...'')
b. 半自动路由


from django.conf.urls import url, include
from web.views import s10_generic
urlpatterns = [
url(r''^test/$'', s10_generic.UserViewSet.as_view({''get'': ''list'', ''post'': ''create''})),
url(r''^test/(?P<pk>\d+)/$'', s10_generic.UserViewSet.as_view(
{''get'': ''retrieve'', ''put'': ''update'', ''patch'': ''partial_update'', ''delete'': ''destroy''})),
]


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.viewsets import ModelViewSet
from rest_framework import serializers
from .. import models
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = models.UserInfo
fields = "__all__"
class UserViewSet(ModelViewSet):
queryset = models.UserInfo.objects.all()
serializer_class = UserSerializer
c. 全自动路由


from django.conf.urls import url, include
from rest_framework import routers
from web.views import s10_generic
router = routers.DefaultRouter()
router.register(r''users'', s10_generic.UserViewSet)
urlpatterns = [
url(r''^'', include(router.urls)),
]


from rest_framework.viewsets import ModelViewSet
from rest_framework import serializers
from .. import models
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = models.UserInfo
fields = "__all__"
class UserViewSet(ModelViewSet):
queryset = models.UserInfo.objects.all()
serializer_class = UserSerializer
9. 视图
a. GenericViewSet


from django.conf.urls import url, include
from web.views.s7_viewset import TestView
urlpatterns = [
url(r''test/'', TestView.as_view({''get'':''list''}), name=''test''),
url(r''detail/(?P<pk>\d+)/'', TestView.as_view({''get'':''list''}), name=''xxxx''),
]


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework import viewsets
from rest_framework.response import Response
class TestView(viewsets.GenericViewSet):
def list(self, request, *args, **kwargs):
return Response(''...'')
def add(self, request, *args, **kwargs):
pass
def delete(self, request, *args, **kwargs):
pass
def edit(self, request, *args, **kwargs):
pass
b. ModelViewSet(自定义URL)


from django.conf.urls import url, include
from web.views import s10_generic
urlpatterns = [
url(r''^test/$'', s10_generic.UserViewSet.as_view({''get'': ''list'', ''post'': ''create''})),
url(r''^test/(?P<pk>\d+)/$'', s10_generic.UserViewSet.as_view(
{''get'': ''retrieve'', ''put'': ''update'', ''patch'': ''partial_update'', ''delete'': ''destroy''})),
]


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.viewsets import ModelViewSet
from rest_framework import serializers
from .. import models
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = models.UserInfo
fields = "__all__"
class UserViewSet(ModelViewSet):
queryset = models.UserInfo.objects.all()
serializer_class = UserSerializer
c. ModelViewSet(rest framework路由)


from django.conf.urls import url, include
from rest_framework import routers
from app01 import views
router = routers.DefaultRouter()
router.register(r''users'', views.UserViewSet)
router.register(r''groups'', views.GroupViewSet)
# Wire up our API using automatic URL routing.
# Additionally, we include login URLs for the browsable API.
urlpatterns = [
url(r''^'', include(router.urls)),
]


from rest_framework import viewsets
from rest_framework import serializers
class UserSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = models.User
fields = (''url'', ''username'', ''email'', ''groups'')
class GroupSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = models.Group
fields = (''url'', ''name'')
class UserViewSet(viewsets.ModelViewSet):
"""
API endpoint that allows users to be viewed or edited.
"""
queryset = User.objects.all().order_by(''-date_joined'')
serializer_class = UserSerializer
class GroupViewSet(viewsets.ModelViewSet):
"""
API endpoint that allows groups to be viewed or edited.
"""
queryset = Group.objects.all()
serializer_class = GroupSerializer
10. 渲染器
根据 用户请求URL 或 用户可接受的类型,筛选出合适的 渲染组件。
用户请求URL:
- http://127.0.0.1:8000/test/?format=json
- http://127.0.0.1:8000/test.json
用户请求头:
- Accept:text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8
a. json
访问URL:
- http://127.0.0.1:8000/test/?format=json
- http://127.0.0.1:8000/test.json
- http://127.0.0.1:8000/test/


from django.conf.urls import url, include
from web.views import s11_render
urlpatterns = [
url(r''^test/$'', s11_render.TestView.as_view()),
url(r''^test\.(?P<format>[a-z0-9]+)'', s11_render.TestView.as_view()),
]


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import serializers
from rest_framework.renderers import JSONRenderer
from .. import models
class TestSerializer(serializers.ModelSerializer):
class Meta:
model = models.UserInfo
fields = "__all__"
class TestView(APIView):
renderer_classes = [JSONRenderer, ]
def get(self, request, *args, **kwargs):
user_list = models.UserInfo.objects.all()
ser = TestSerializer(instance=user_list, many=True)
return Response(ser.data)
b. 表格
访问URL:
- http://127.0.0.1:8000/test/?format=admin
- http://127.0.0.1:8000/test.admin
- http://127.0.0.1:8000/test/


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import serializers
from rest_framework.renderers import AdminRenderer
from .. import models
class TestSerializer(serializers.ModelSerializer):
class Meta:
model = models.UserInfo
fields = "__all__"
class TestView(APIView):
renderer_classes = [AdminRenderer, ]
def get(self, request, *args, **kwargs):
user_list = models.UserInfo.objects.all()
ser = TestSerializer(instance=user_list, many=True)
return Response(ser.data)
c. Form表单
访问URL:
- http://127.0.0.1:8000/test/?format=form
- http://127.0.0.1:8000/test.form
- http://127.0.0.1:8000/test/


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import serializers
from rest_framework.renderers import JSONRenderer
from rest_framework.renderers import AdminRenderer
from rest_framework.renderers import HTMLFormRenderer
from .. import models
class TestSerializer(serializers.ModelSerializer):
class Meta:
model = models.UserInfo
fields = "__all__"
class TestView(APIView):
renderer_classes = [HTMLFormRenderer, ]
def get(self, request, *args, **kwargs):
user_list = models.UserInfo.objects.all().first()
ser = TestSerializer(instance=user_list, many=False)
return Response(ser.data)
d. 自定义显示模板
访问URL:
- http://127.0.0.1:8000/test/?format=html
- http://127.0.0.1:8000/test.html
- http://127.0.0.1:8000/test/


from django.conf.urls import url, include
from web.views import s11_render
urlpatterns = [
url(r''^test/$'', s11_render.TestView.as_view()),
url(r''^test\.(?P<format>[a-z0-9]+)'', s11_render.TestView.as_view()),
]


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import serializers
from rest_framework.renderers import TemplateHTMLRenderer
from .. import models
class TestSerializer(serializers.ModelSerializer):
class Meta:
model = models.UserInfo
fields = "__all__"
class TestView(APIView):
renderer_classes = [TemplateHTMLRenderer, ]
def get(self, request, *args, **kwargs):
user_list = models.UserInfo.objects.all().first()
ser = TestSerializer(instance=user_list, many=False)
return Response(ser.data, template_name=''user_detail.html'')


<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
{{ user }}
{{ pwd }}
{{ ut }}
</body>
</html>
e. 浏览器格式API+JSON
访问URL:
- http://127.0.0.1:8000/test/?format=api
- http://127.0.0.1:8000/test.api
- http://127.0.0.1:8000/test/


#!/usr/bin/env python
# -*- coding:utf-8 -*-
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import serializers
from rest_framework.renderers import JSONRenderer
from rest_framework.renderers import BrowsableAPIRenderer
from .. import models
class TestSerializer(serializers.ModelSerializer):
class Meta:
model = models.UserInfo
fields = "__all__"
class CustomBrowsableAPIRenderer(BrowsableAPIRenderer):
def get_default_renderer(self, view):
return JSONRenderer()
class TestView(APIView):
renderer_classes = [CustomBrowsableAPIRenderer, ]
def get(self, request, *args, **kwargs):
user_list = models.UserInfo.objects.all().first()
ser = TestSerializer(instance=user_list, many=False)
return Response(ser.data, template_name=''user_detail.html'')
注意:如果同时多个存在时,自动根据URL后缀来选择渲染器。
Django rest framework (4) --- 版本
版本
新建一个工程 Myproject 和一个 app 名为 api
(1)api/models.py

from django.db import models
class UserInfo(models.Model):
USER_TYPE = (
(1,''普通用户''),
(2,''VIP''),
(3,''SVIP'')
)
user_type = models.IntegerField(choices=USER_TYPE)
username = models.CharField(max_length=32,unique=True)
password = models.CharField(max_length=64)
group = models.ForeignKey(''UserGroup'',on_delete=models.CASCADE)
roles = models.ManyToManyField(''Role'')
class UserToken(models.Model):
user = models.OneToOneField(''UserInfo'',on_delete=models.CASCADE)
token = models.CharField(max_length=64)
class UserGroup(models.Model):
title = models.CharField(max_length=32)
class Role(models.Model):
title = models.CharField(max_length=32)

(2)Myproject/urls.py

from django.conf.urls import url,include
from django.contrib import admin
urlpatterns = [
# url(r''^admin/'', admin.site.urls),
url(r''^api/'', include(''api.urls'')),
]

(3)api/urls.py

# api/urls.py
from django.conf.urls import url,include
from django.contrib import admin
from api.views import UserView
urlpatterns = [
# url(r''^admin/'', admin.site.urls),
url(r''^users/'', UserView.as_view()),
]

(4)views.py

# api/views.py
from django.shortcuts import render,HttpResponse
from rest_framework.views import APIView
from rest_framework.request import Request
from rest_framework.versioning import QueryParameterVersioning
class UserView(APIView):
versioning_class = QueryParameterVersioning
def get(self,request,*args,**kwargs):
#获取版本
print(request.version)
return HttpResponse(''用户列表'')

(5)settings.py

#版本
REST_FRAMEWORK = {
"DEFAULT_VERSION":''v1'', #默认的版本
"ALLOWED_VERSIONS":[''v1'',''v2''], #允许的版本
"VERSION_PARAM":''version'' #GET方式url中参数的名字 ?version=xxx
}

1.url 中通过 GET 传参
QueryParameterVersioning用于去GET参数中取version
http://127.0.0.1:8000/api/users/?version=v2
后台可以看到当前的版本
如果 url 中没有传版本参数,则显示默认的版本("DEFAULT_VERSION":''v1'')
http://127.0.0.1:8000/api/users/
如果 url 传的版本超过 settings 中的允许范围则报错
http://127.0.0.1:8000/api/users/?version=v3
2. 在 URLPATH 中获取
(1)修改 api/urls.py
通常情况我门应该用 URLPATH 的方式,而不是用前面 GET () 传参方式
url 里面通过正则表达式定义哪些版本,

from django.conf.urls import url,include
from django.contrib import admin
from api.views import UserView
urlpatterns = [
# url(r''^admin/'', admin.site.urls),
# url(r''^users/'', UserView.as_view()),
url(''(?P<version>[v1|v2]+)/users/'', UserView.as_view()),
]
(2)views.py
URLPathVersioning:去url路径里面获取版本

# api/views.py
from django.shortcuts import render,HttpResponse
from rest_framework.views import APIView
from rest_framework.request import Request
from rest_framework.versioning import URLPathVersioning
class UserView(APIView):
versioning_class = URLPathVersioning
def get(self,request,*args,**kwargs):
#获取版本
print(request.version)
return HttpResponse(''用户列表'')

这个 URLPathVersioning 我们可以放到 settings 里面,全局配置,就不用写到 views 里面,每个类都要写一遍了
settings.py

# 版本
# REST_FRAMEWORK = {
# "DEFAULT_VERSIONING_CLASS":"rest_framework.versioning.URLPathVersioning",
# "DEFAULT_VERSION":''v1'', #默认的版本
# "ALLOWED_VERSIONS":[''v1'',''v2''], #允许的版本
# "VERSION_PARAM":''version'' #get方式url中参数的名字 ?version=xxx
# }
#全局
REST_FRAMEWORK = {
"DEFAULT_VERSIONING_CLASS":"rest_framework.versioning.URLPathVersioning",
}

修改 views.py

# api/views.py
from django.shortcuts import render,HttpResponse
from rest_framework.views import APIView
from rest_framework.request import Request
class UserView(APIView):
def get(self,request,*args,**kwargs):
#获取版本
print(request.version)
return HttpResponse(''用户列表'')

浏览器访问地址
http://127.0.0.1:8000/api/v1/users/
然后后台拿到版本信息
3. 反向解析访问的 url
1)api/urls.py
添加 name = ''api_user''

# api/urls.py
from django.conf.urls import url,include
from django.contrib import admin
from api.views import UserView
urlpatterns = [
url(''(?P<version>[v1|v2]+)/users/'', UserView.as_view(),name=''api_user''),
]

(2)views.py

# api/views.py
from django.shortcuts import render,HttpResponse
from rest_framework.views import APIView
from rest_framework.request import Request
class UserView(APIView):
def get(self,request,*args,**kwargs):
#获取版本
print(request.version)
#获取处理版本的对象
print(request.versioning_scheme)
#获取浏览器访问的url,reverse反向解析
#需要两个参数:viewname就是url中的别名,request=request是url中要传入的参数
#(?P<version>[v1|v2]+)/users/,这里本来需要传version的参数,但是version包含在request里面(源码里面可以看到),所有只需要request=request就可以
url_path = request.versioning_scheme.reverse(viewname=''api_user'',request=request)
print(url_path)
# self.dispatch
return HttpResponse(''用户列表'')

浏览器访问
http://127.0.0.1:8000/api/v1/users/
后台获取
源码流程
(1)dispatch

def dispatch(self, request, *args, **kwargs):
"""
`.dispatch()` is pretty much the same as Django''s regular dispatch,
but with extra hooks for startup, finalize, and exception handling.
"""
self.args = args
self.kwargs = kwargs
#对原始request进行加工,丰富了一些功能
#Request(
# request,
# parsers=self.get_parsers(),
# authenticators=self.get_authenticators(),
# negotiator=self.get_content_negotiator(),
# parser_context=parser_context
# )
#request(原始request,[BasicAuthentications对象,])
#获取原生request,request._request
#获取认证类的对象,request.authticators
#1.封装request
request = self.initialize_request(request, *args, **kwargs)
self.request = request
self.headers = self.default_response_headers # deprecate?
try:
#2.认证
self.initial(request, *args, **kwargs)
# Get the appropriate handler method
if request.method.lower() in self.http_method_names:
handler = getattr(self, request.method.lower(),
self.http_method_not_allowed)
else:
handler = self.http_method_not_allowed
response = handler(request, *args, **kwargs)
except Exception as exc:
response = self.handle_exception(exc)
self.response = self.finalize_response(request, response, *args, **kwargs)
return self.response

(2)initial

def initial(self, request, *args, **kwargs):
"""
Runs anything that needs to occur prior to calling the method handler.
"""
self.format_kwarg = self.get_format_suffix(**kwargs)
# Perform content negotiation and store the accepted info on the request
neg = self.perform_content_negotiation(request)
request.accepted_renderer, request.accepted_media_type = neg
# Determine the API version, if versioning is in use.
#request.version获取版本信息
#request.versioning_scheme获取处理版本你的对象
version, scheme = self.determine_version(request, *args, **kwargs)
request.version, request.versioning_scheme = version, scheme
# Ensure that the incoming request is permitted
#4.实现认证
self.perform_authentication(request)
#5.权限判断
self.check_permissions(request)
#6.控制访问频率
self.check_throttles(request)

(3)determine_version

def determine_version(self, request, *args, **kwargs):
"""
If versioning is being used, then determine any API version for the
incoming request. Returns a two-tuple of (version, versioning_scheme)
"""
if self.versioning_class is None:
return (None, None)
scheme = self.versioning_class()
return (scheme.determine_version(request, *args, **kwargs), scheme)

(4)versioning_class
URLPathVersioning 源码

class URLPathVersioning(BaseVersioning):
"""
To the client this is the same style as `NamespaceVersioning`.
The difference is in the backend - this implementation uses
Django''s URL keyword arguments to determine the version.
An example URL conf for two views that accept two different versions.
urlpatterns = [
url(r''^(?P<version>[v1|v2]+)/users/$'', users_list, name=''users-list''),
url(r''^(?P<version>[v1|v2]+)/users/(?P<pk>[0-9]+)/$'', users_detail, name=''users-detail'')
]
GET /1.0/something/ HTTP/1.1
Host: example.com
Accept: application/json
"""
invalid_version_message = _(''Invalid version in URL path.'')
def determine_version(self, request, *args, **kwargs):
version = kwargs.get(self.version_param, self.default_version)
if not self.is_allowed_version(version):
raise exceptions.NotFound(self.invalid_version_message)
return version
def reverse(self, viewname, args=None, kwargs=None, request=None, format=None, **extra):
if request.version is not None:
kwargs = {} if (kwargs is None) else kwargs
kwargs[self.version_param] = request.version
return super(URLPathVersioning, self).reverse(
viewname, args, kwargs, request, format, **extra
)

可以看到
(1)url 配置
(2)determine_version
里面有个 is_allowed_version,点进去可以看到一些基本参数 (继承 BaseVersioning 基类)

class BaseVersioning(object):
#默认的版本
default_version = api_settings.DEFAULT_VERSION
#允许的版本
allowed_versions = api_settings.ALLOWED_VERSIONS
#默认参数(是version,比如你可以自定义为v)
version_param = api_settings.VERSION_PARAM
def determine_version(self, request, *args, **kwargs):
msg = ''{cls}.determine_version() must be implemented.''
raise NotImplementedError(msg.format(
cls=self.__class__.__name__
))
def reverse(self, viewname, args=None, kwargs=None, request=None, format=None, **extra):
return _reverse(viewname, args, kwargs, request, format, **extra)
def is_allowed_version(self, version):
if not self.allowed_versions:
return True
return ((version is not None and version == self.default_version) or
(version in self.allowed_versions))
关于Django Rest Framework部分更新和django 局部更新数据的问题我们已经讲解完毕,感谢您的阅读,如果还想了解更多关于1 Django REST Framework 开发 ---- 原生 Django View 热身、Django - rest - framework - 下、Django Rest Framework、Django rest framework (4) --- 版本等相关内容,可以在本站寻找。
本文标签: