在这篇文章中,我们将为您详细介绍Djangodrf:幂等性的内容,并且讨论关于幂等性redis的相关问题。此外,我们还会涉及一些关于2020-05-12:哪些场景需要幂等性?怎么保证幂等性?、Djan
在这篇文章中,我们将为您详细介绍Django drf:幂等性的内容,并且讨论关于幂等性 redis的相关问题。此外,我们还会涉及一些关于2020-05-12:哪些场景需要幂等性?怎么保证幂等性?、Django -- DRF 认证流程、Django 2.0 学习(22):Django CSRF、django DRF的知识,以帮助您更全面地了解这个主题。
本文目录一览:- Django drf:幂等性(幂等性 redis)
- 2020-05-12:哪些场景需要幂等性?怎么保证幂等性?
- Django -- DRF 认证流程
- Django 2.0 学习(22):Django CSRF
- django DRF
Django drf:幂等性(幂等性 redis)
一、什么叫做幂等性
用户对于同一操作发起的一次请求或者多次请求的结果是一致的,不会因为多次点击而产生了副作用。举个最简单的例子,那就是支付,用户购买商品使用约支付,支付扣款成功,但是返回结果的时候网络异常,此时钱已经扣了,用户再次点击按钮,此时会进行第二次扣款,返回结果成功,用户查询余额返发现多扣钱了,流水记录也变成了两条...
在以前的单应用系统中,我们只需要把数据操作放入事务中即可,发生错误立即回滚,但是再响应客户端的时候也有可能出现网络中断或者异常等等。
二、post 添加数据不具有幂等性
在增删改查 4 个操作中,尤为注意就是增加或者修改,
查询对于结果是不会有改变的,(get)
删除只会进行一次,用户多次点击产生的结果一样(delete)
修改在大多场景下结果一样(put)
增加在重复提交的场景下会出现(post)
三、解决方案
方法一:ticketId
单次支付请求,也就是直接支付了,不需要额外的数据库操作了,这个时候发起异步请求创建一个唯一的ticketId,就是门票,这张门票只能使用一次就作废,具体步骤如下:
1.异步请求获取门票
2.调用支付,传入门票
3.根据门票ID查询此次操作是否存在,如果存在则表示该操作已经执行过,直接返回结果;如果不存在,支付扣款,保存结果
4.返回结果到客户端
如果步骤4通信失败,用户再次发起请求,那么最终结果还是一样的
方法二:方法二:分布式环境下各个服务相互调用
这边就要举例我们的系统了,我们支付的时候先要扣款,然后更新订单,这个地方就涉及到了订单服务以及支付服务了。
用户调用支付,扣款成功后,更新对应订单状态,然后再保存流水。
而在这个地方就没必要使用门票ticketId了,因为会比较闲的麻烦
(支付状态:未支付,已支付)
步骤:
1、查询订单支付状态
2、如果已经支付,直接返回结果
3、如果未支付,则支付扣款并且保存流水
4、返回支付结果
如果步骤4通信失败,用户再次发起请求,那么最终结果还是一样的
对于做过支付的朋友,幂等,也可以称之为冲正,保证客户端与服务端的交易一致性,避免多次扣款。
2020-05-12:哪些场景需要幂等性?怎么保证幂等性?
答案仅供参考:
2.需要幂等的场景
可能会发生重复请求或消费的场景,在微服务架构中是随处可见的。以下是笔者梳理的几个常见场景:
网络波动:
因网络波动,可能会引起重复请求
分布式消息消费:
任务发布后,使用分布式消息服务来进行消费,参考【消息总线真的能保证幂等?】
用户重复操作:
用户在使用产品时,可能会误操作而触发多笔交易,或者因为长时间没有响应,而有意触发多笔交易。
未关闭的重试机制:
技术人员人为的错误,因开发人员、测试人员或运维人员没有检查出来,而开启的重试机制(如Nginx重试、RPC通信重试或业务层重试等)
4.幂等实现方法:
4.1 全局唯一ID
如果使用全局唯一ID,就是根据业务的操作和内容生成一个全局ID,在执行操作前先根据这个全局唯一ID是否存在,来判断这个操作是否已经执行。如果不存在则把全局ID,存储到存储系统中,比如数据库、Redis等。如果存在则表示该方法已经执行。
使用全局唯一ID是一个通用方案,可以支持插入、更新、删除业务操作。但是这个方案看起来很美但是实现起来比较麻烦,下面的方案适用于特定的场景,但是实现起来比较简单。
4.2 去重表
这种方法适用于在业务中有唯一标的插入场景中,比如在以上的支付场景中,如果一个订单只会支付一次,所以订单ID可以作为唯一标识。这时,我们就可以建一张去重表,并且把唯一标识作为唯一索引,用以记录订单支付信息,在我们实现时,把创建支付单据和写入去去重表,放在一个事务中,如果重复创建,数据库会抛出唯一约束异常,操作就会回滚。这个方法其实也是用到唯一ID,与上面全局唯一ID不同的是,他是针对具体单个业务流程的,实现起来相对简单。
4.3 插入或更新
这种方法插入并且有唯一索引的情况,比如我们要关联商品品类,其中商品的ID和品类的ID可以构成唯一索引,并且在数据表中也增加了唯一索引。这时就可以使用InsertOrUpdate操作。在mysql数据库中如下:
insert into goods_category
(goods_id,category_id,create_time,update_time)
values(#{goodsId},#{categoryId},now(),now())
on DUPLICATE KEY UPDATE update_time=now()
4.4 多版本控制
这种方法适合在更新的场景中,比如我们要更新商品的名字,这时我们就可以在更新的接口中增加一个版本号,来做幂等:
boolean updateGoodsName(int id,String newName,int version);
在实现时可以如下:
update goods set name=#{newName},version=#{version} where
id=#{id} and version<${version}
4.5 状态机控制
这种方法适合在有状态机流转的情况下,比如就会订单的创建和付款,订单的付款肯定是在之前,这时我们可以通过在设计状态字段时,使用int类型,并且通过值类型的大小来做幂等,比如订单的创建为0,付款成功为100,付款失败为99。在做状态机更新时,我们就这可以这样控制:
update goods_order set status=#{status} where id=#{id} and
status<#{status}
以上就是保证接口幂等性的一些方法。
5.总结
幂等性设计不能脱离业务来讨论,一般情况下,去重表同时也是业务数据表,而针对分布式的去重ID,可以参考以下几种方式:
UUID
Snowflake
数据库自增ID
业务本身的唯一约束
业务字段+时间戳拼接
评论
Django -- DRF 认证流程
Django Restful Framework (DRF) 中类的调用与自定义 -- 以 autentication 认证为例
<br>
DRF 的 request 对 django 的 request 进行了更一步的封装;通过获取认证相关的所有类,并实例化,传入 request 对象 (user,auth)
<br>
自定义实例
from django.shortcuts import render
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.request import Request
from goods.serializers import GoodsSerializer
from rest_framework.exceptions import APIException
# Create your views here.
from goods.models import Goods
class MyAuthentication(object):
def authenticate(self,request):
token = request.query_params.get(''token'')
if token == ''abc'':
return (''aaa'',''bbb'')
raise APIException(''认证失败'')
class GoodsListView(APIView):
authentication_classes = [MyAuthentication,]
def get(self,request,*args,**kwargs):
print(request.user,request.auth)
goods = Goods.objects.all()
goods_serializer = GoodsSerializer(goods,many=True)
return Response(goods_serializer.data)
<br>
流程
<br>
在使用的时候 使用 BaseAuthentication 类
<br>
这里可以把认证 当成用户登录状态的判断
登录时不需要 认证,登录之后的访问都需要 进行 登录用户身份的确认。
可以全局添加认证,然后把登录,或者主页等不需要 登录的视图,设置为 authentication_classes= []
<br>
当 authenticate 不处理返回 None 的时候,匿名用户的设置
REST_FRAMEWORK = {
"DEFAULT_AUTHENTICATION_CLASSES":[],
# "UNAUTHENTICATED_USER":lambda :''匿名用户'',
"UNAUTHENTICATED_USER":None,
# "UNAUTHENTICATED_TOKEN":lambda :''没有验证信息'',
"UNAUTHENTICATED_TOKEN":None,
}
<br>
BasicAuthentication
def authenticate_header(self, request):
return ''Basic realm="%s"'' % ''api
Django 2.0 学习(22):Django CSRF
Django CSRF
CSRF攻击过程
攻击说明: 1.用户C打开浏览器,访问受信任网站A,输入用户名和密码请求登陆网站A; 2.在用户信息通过验证后,网站A产生Cookie信息并返回给浏览器,此时用户登陆网站A成功,可以正常发送请求到网站A; 3.用户未退出网站A之前,在同一浏览器中,打开一个TAB页访问网站B; 4.网站B收到用户请求后,返回一些攻击性代码,并发出一个请求,要求访问第三方站点A; 5.浏览器在接收到这些攻击性代码后,根据网站B的请求,在用户不知情的情况下携带Cookie信息,向网站A发出请求。网站A并不知道该请求其实是由B发起的,所以会根据用户C的Cookie信息以C的权限处理该请求,导致来自网站B的恶意代码被执行。
CSRF的攻击之所以会成功是因为服务器端身份验证机制可以通过Cookie保证一个请求是来自于某个用户的浏览器,>单无法保证该请求是用户允许的。因此,预防CSRF攻击简单可行的方法就是在客户端网页上添加随机数,在服务器>端进行随机数验证,以确保该请求是用户允许的。Django也是通过这个方法来预防CSRF攻击的。
Django防御CSRF攻击
原理 在客户端页面上天机csrftoken,服务器端进行验证。服务器端验证的工作通过"django.middleware.csrf.CsrfViewMiddleware"这个中间件来完成。在Django中防御csrf攻击的方式有两种:
- 在表单中附加csrftoken;
- 通过request请求中添加x-csrftoken请求头;
**注意:Django默认对所有的POST请求都进行csrftoken验证,若验证失败则返回403错误。**Django中设置防跨站请求伪造功能分为全局和局部。 全局:中间件 django.middleware.csrf.CsrfViewMiddleware 局部:from django.views.decorators.csrf import csrf_protect, csrf_exempt
- @csrf_protect,为当前函数强制设置防跨站请求伪造功能,即便settings中没有设置全局中间件;
- @csrf_exempt,取消当前函数防跨站请求伪造功能,即便settings中设置了全局中间件;
当用post提交数据的时候,django会去检查是否有一个csrf的随机字符串,如果没有就会报错,错误如下: 在Django内部支持生成这个随机字符串。
通过form提交 在form表单里面需要添加{% csrf_token %},这样当查看页面源码的时候,可以看到form中有一个input是隐藏的:
原理总结: 当用户访问login页面的时候,会生成已给csrf的随机字符串,并且cookie中野村放了这个随机字符串,当用户再次提交数据的时候会带着这个随机字符串提交,如果没有这个随机字符串则无法提交成功。cookie中存放的csrftoken如下图所示:
通过ajax提交 因为cookie中同样存在csrftoken,所以可以在JavaScript中通过$.cookie("csrftoken")获取。如果通过ajax进行提交数据,这里提交的csrftoken是通过请求头中存放,需要提交一个字典型的数据,即这时候需要一个key。在views中的login函数中:from django.conf import settings,然后打印print(settings.CSRF_HEADER_NAME),这里需要注意一个问题,这里导入的settings并不是我们在项目下看到的settings.py文件,这里是一个全局的settings配置,而当我们在项目目录下的settings.py中配置的时候,我们添加的配置则会覆盖全局settings中的配置。print(settings.CSRF_HEADER_NAME)打印的内容为:HTTP_X_CSRFTOKEN,这里的HTTP_X_CSRFTOEKN是Django在X_CSRF的前面添加了HTTP_,所以实际传递的就是X_CSRFTOKEN,而在前端页面的ajax传递的时候由于不能使用下划线,所以传递的是X_CSRFTOKEN。下面是在前端ajax中写的具体内容:
$("#btn1").click(function () {
$.ajax({
url:"/login/",
type:"POST",
data:{"usr":"root","pwd":"123"},
headers:{ "X-CSRFtoken":$.cookie("csrftoken")},
success:function (arg) {
};
});
});
但是如果页面中有多个ajax请求的话,就在每个ajax中添加headers信息,所以可以通过下面方式在所有的ajax中都添加:
$.ajaxSetup({
beforeSend:function (xhr,settings) {
xhr.setRequestHeader("X-CSRFtoken",$.cookie("csrftoken"))
}
});
这样就会在提交ajax之前执行这个方法,从而在所有的ajax里都加上这个csrftoken,这里的xhr是XMLHttpRequest的简写,ajax调用的就是这个方法。如果想要实现在当get方式的时候不需要提交csrftoken,当post的时候需要,实现这种效果的代码如下:
function csrfSafeMethod(method) {
// these HTTP methods do not require CSRF protection
return (/^(GET|HEAD|OPTIONS|TRACE)$/.test(method));
}
$.ajaxSetup({
beforeSend: function(xhr, settings) {
if (!csrfSafeMethod(settings.type) && !this.crossDomain) {
xhr.setRequestHeader("X-CSRFToken", csrftoken);
};
};
});
这样就实现了当GET|HEAD|OPTIONS|TRACE这些方式请求的时候不需要提交csrftoken。
总结: 1、csrf在ajax提交的时候通过请求头传递给后台的; 2、csrf在前端的key为:X-CSRFToken,到后端的时候Django会自动添加HTTP_,并且最后为HTTP_X_CSRFTOKEN; 3、csrf在form中提交的时候需要在前端form中添加{% csrftoken %};
django DRF
DjangoRestFramework
1. 安装:
pip install djangorestframework
2. restful规范:
2.1 API与用户通信协议,总是使用https协议(我们现在用的大部分都是http)
2.2 域名
https://api.cainiao.com 尽量将API部署在专用域名(会存在跨域问题)
https://cainiao.com/api/ API很简单(推荐使用)
2.3 版本
URL,如:https://api.cainiao.com/v1/
存在跨域问题咱不慌,请往这里看=>:https://www.cnblogs.com/xingxingnbsp/p/10644135.html
2.4 路径,将网络上的所有东西都看成是资源,均使用名词表示(可以使用复数)
https://api.cainiao.com/v1/zoos
https://api.cainiao.com/v1/animls
https://api.cainiao.com/v1/employees
2.5 method(请求方式)
GET : 从服务器取出资源(一项或多项)
POST : 在服务器新建一个资源
PUT : 在服务器更新资源(客户端提供改变后的完整资源)
PATCH : 在服务器更新资源(客户端提供改变的属性)
DELETE : 从服务器删除资源
2.6 过滤,通过在url上传参的形式传递搜索条件
https://api.cainiao.com/v1/zoos?limit=10 : 指定返回记录的数量
https://api.cainiao.com/v1/zoos?offset=10 : 指定返回记录的数量
https://api.cainiao.com/v1/zoos?page=2&per_page=100 : 指定第几页,以及每页的记录数
https://api.cainiao.com/v1/zoos?sortby=name&order=asc : 指定返回结果按照哪个属性排序,以及排序规则
2.7 状态码
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
2.8 错误处理
{
error:"lnvalid API key"
}
2.9 返回结果,针对不同操作,服务器向用户返回的结果应该符合以下规范。
GET/collection : 返回资源对象的列表(数组)
GET/collection/response : 返回单个资源对象
POST/collection : 返回新生成的资源对象
PUT/collection/response : 返回完整的资源对象
PATCH/collection/response :返回完整的资源对象
DELETE/collection/response :返回一个空文档
2.10 Hypermedia API,RESTful API最好做到Hypermedia,即返回结果中提供链接,连向其他API方法,使得用户不查文档,也知道下一步应该做什么。
{"link": {
"rel": "collection https://www.cainiao.com/zoos",
"href": "https://api.cainiao.com/zoos",
"title": "List of zoos",
"type": "application/vnd.yourformat+json"
}}
3. 配置文件(settings.py)
# django rest framework配置
REST_FRAMEWORK = {
# 解析器配置
''DEFAULT_PARSER_CLASSES'':[
''rest_framework.parsers.JSONParser''
''rest_framework.parsers.FormParser''
''rest_framework.parsers.MultiPartParser''
]
# 版本配置
''DEFAULT_VERSIONING_CLASS'':"rest_framework.versioning.URLPathVersioning",
''DEFAULT_VERSION'': ''v1'', # 默认版本
''ALLOWED_VERSIONS'': [''v1'', ''v2''], # 允许的版本
''VERSION_PARAM'': ''version'' # URL中获取值的key
# 用户登录认证
''UNAUTHENTICATED_USER'': None,
''UNAUTHENTICATED_TOKEN'': None,
"DEFAULT_AUTHENTICATION_CLASSES": ["web.utils.TestAuthentication",],
# 用户权限验证
"DEFAULT_PERMISSION_CLASSES": ["web.utils.TestPermission",],
# 用户访问频率限制(匿名时用IP限制+登录时用Token限制)
''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.模块详细介绍:
4.1 认证模块,权限模块:
4.1.1 用户url传入的token认证:
urls.py:
from django.conf.urls import url, include
from web.viewsimport TestView
urlpatterns = [
url(r''^test/'', TestView.as_view()),
]
views.py:
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):
"""
注释001
"""
val = request.query_params.get(''token'')
if val not in token_list:
raise exceptions.AuthenticationFailed("用户认证失败")
return (''登录用户'', ''用户token'')
def authenticate_header(self, request):
# 验证失败时,返回的响应头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请求,响应内容'')
4.1.2 请求头认证:
urls.py:
from django.conf.urls import url, include
from web.viewsimport TestView
urlpatterns = [
url(r''^test/'', TestView.as_view()),
]
views.py:
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):
"""
注释001
"""
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 ''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请求,响应内容'')
4.1.3 多个认证规则:
urls.py:
from django.conf.urls import url, include
from web.views.s2_auth import TestView
urlpatterns = [
url(r''^test/'', TestView.as_view()),
]
views.py:
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):
"""
注释001
"""
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 ''Basic realm=api''
pass
class Test2Authentication(BaseAuthentication):
def authenticate(self, request):
"""
注释001
"""
val = request.query_params.get(''token'')
if val not in token_list:
raise exceptions.AuthenticationFailed("用户认证失败")
return (''登录用户'', ''用户token'')
def authenticate_header(self, request):
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请求,响应内容'')
4.1.4 认证和权限:
urls.py:
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''^test/'', TestView.as_view()),
]
views.py:
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):
"""
注释001
"""
val = request.query_params.get(''token'')
if val not in token_list:
raise exceptions.AuthenticationFailed("用户认证失败")
return (''登录用户'', ''用户token'')
def authenticate_header(self, request):
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请求,响应内容'')
4.1.5 全局使用:
settings.py:
REST_FRAMEWORK = {
''UNAUTHENTICATED_USER'': None,
''UNAUTHENTICATED_TOKEN'': None,
"DEFAULT_AUTHENTICATION_CLASSES": [
"web.utils.TestAuthentication",
],
"DEFAULT_PERMISSION_CLASSES": [
"web.utils.TestPermission",
],
}
urls.py:
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''^test/'', TestView.as_view()),
]
views.py:
from rest_framework.views import APIView
from rest_framework.response import Response
class TestView(APIView):
def 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请求,响应内容'')
4.1.6 基于django-rest-framework的JWT认证(推荐使用):
安装:
pip install djangorestframework-jwt
配置:
REST_FRAMEWORK = {
''DEFAULT_AUTHENTICATION_CLASSES'': (
''rest_framework_jwt.authentication.JSONWebTokenAuthentication'', #新增
''rest_framework.authentication.SessionAuthentication'',
''rest_framework.authentication.BasicAuthentication'',
),
}
JWT_AUTH = {
#JWT_EXPIRATION_DELTA 指明token的有效期
''JWT_EXPIRATION_DELTA'':datetime.timedelta(days=1),
}
使用:
from rest_framework_jwt.settings import api_settings
jwt_payload_handler = api_settings.JWT_PAYLOAD_HANDLER
jwt_encode_handler = api_settings.JWT_ENCODE_HANDLER
payload = jwt_payload_handler(user)
token = jwt_encode_handler(payload)
使用Django REST framework JWT进行用户登录:
from rest_framework_jwt.views import obtain_jwt_token
urlpatterns = [
url(r''^authorizations/$'', obtain_jwt_token),
]
4.2 用户访问次数/频率限制:
4.2.1 基于用户IP限制访问频率:
urls.py
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''^test/'', TestView.as_view()),
]
views.py:
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,当做请求者的唯一标识
"""
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)
4.2.2 基于用户IP显示访问频率(利于Django缓存):
settings.py:
REST_FRAMEWORK = {
''DEFAULT_THROTTLE_RATES'': {
''test_scope'': ''10/m'',
},
}
urls.py:
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''^test/'', TestView.as_view()),
]
views.py:
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):
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)
4.2.3 view中限制请求频率:
settings.py:
REST_FRAMEWORK = {
''DEFAULT_THROTTLE_RATES'': {
''xxxxxx'': ''10/m'',
},
}
urls.py:
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''^test/'', TestView.as_view()),
]
views.py:
# !/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):
"""
应该返回一个惟一的缓存键,该缓存键可用于节流。
必须覆盖。
如果请求不应该被节流,可以返回“None”。
"""
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)
4.3.4 匿名时用IP限制+登录时用Token限制:
settings.py:
REST_FRAMEWORK = {
''UNAUTHENTICATED_USER'': None,
''UNAUTHENTICATED_TOKEN'': None,
''DEFAULT_THROTTLE_RATES'': {
''luffy_anon'': ''10/m'',
''luffy_user'': ''20/m'',
},
}
urls.py:
from django.conf.urls import url, include
from web.views.s3_throttling import TestView
urlpatterns = [
url(r''^test/'', TestView.as_view()),
]
views.py:
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请求,响应内容'')
4.3.5 全局使用
settings.py:
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.3 版本控制:
4.3.1 基于url的get传参方式(如:/users?version=v1):
settings.py:
REST_FRAMEWORK = {
''DEFAULT_VERSION'': ''v1'', # 默认版本
''ALLOWED_VERSIONS'': [''v1'', ''v2''], # 允许的版本
''VERSION_PARAM'': ''version'' # URL中获取值的key
}
urls.py:
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''^test/'', TestView.as_view(), name=''test''),
]
views.py:
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请求,响应内容'')
4.3.2 基于url的正则方式(如:/v1/users/):
settings.py:
REST_FRAMEWORK = {
''DEFAULT_VERSION'': ''v1'', # 默认版本
''ALLOWED_VERSIONS'': [''v1'', ''v2''], # 允许的版本
''VERSION_PARAM'': ''version'' # URL中获取值的key
}
urls.py:
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''),
]
views.py:
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请求,响应内容'')
4.3.3 基于 accept 请求头方式(如:Accept: application/json; version=1.0):
settings.py:
REST_FRAMEWORK = {
''DEFAULT_VERSION'': ''v1'', # 默认版本
''ALLOWED_VERSIONS'': [''v1'', ''v2''], # 允许的版本
''VERSION_PARAM'': ''version'' # URL中获取值的key
}
urls.py:
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''^test/'', TestView.as_view(), name=''test''),
]
views.py:
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请求,响应内容'')
4.3.4 基于主机名方法(如:v1.example.com):
settings.py:
ALLOWED_HOSTS = [''*'']
REST_FRAMEWORK = {
''DEFAULT_VERSION'': ''v1'', # 默认版本
''ALLOWED_VERSIONS'': [''v1'', ''v2''], # 允许的版本
''VERSION_PARAM'': ''version'' # URL中获取值的key
}
urls.py:
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''^test/'', TestView.as_view(), name=''test''),
]
views.py:
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请求,响应内容'')
4.3.5 基于django路由系统的namespace(如:example.com/v1/users/):
settings.py:
REST_FRAMEWORK = {
''DEFAULT_VERSION'': ''v1'', # 默认版本
''ALLOWED_VERSIONS'': [''v1'', ''v2''], # 允许的版本
''VERSION_PARAM'': ''version'' # URL中获取值的key
}
urls.py:
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'')),
]
views.py:
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请求,响应内容'')
4.3.6 全局使用:
settings.py:
REST_FRAMEWORK = {
''DEFAULT_VERSIONING_CLASS'':"rest_framework.versioning.URLPathVersioning",
''DEFAULT_VERSION'': ''v1'',
''ALLOWED_VERSIONS'': [''v1'', ''v2''],
''VERSION_PARAM'': ''version''
}
4.4 解析器(parser):(根据请求头 content-type 选择对应的解析器就请求体内容进行处理。)
4.4.1 仅处理请求头content-type为application/json的请求体:
urls.py:
from django.conf.urls import url, include
from web.views.s5_parser import TestView
urlpatterns = [
url(r''test/'', TestView.as_view(), name=''test''),
]
views.py:
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请求,响应内容'')
4.4.2 仅处理请求头content-type为application/x-www-form-urlencoded 的请求体:
urls.py:
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''test/'', TestView.as_view(), name=''test''),
]
views.py:
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请求,响应内容'')
4.4.3 仅处理请求头content-type为multipart/form-data的请求体
urls.py:
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''test/'', TestView.as_view(), name=''test''),
]
views.py:
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请求,响应内容'')
upload.html:
<!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>
4.4.4 仅上传文件:
urls.py:
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''test/(?P<filename>[^/]+)'', TestView.as_view(), name=''test''),
]
views.py:
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请求,响应内容'')
upload.html:
<!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>
4.4.5 同时多个Parser(当同时使用多个parser时,rest framework会根据请求头content-type自动进行比对,并使用对应parser)
urls.py:
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''test/'', TestView.as_view(), name=''test''),
]
views.py:
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请求,响应内容'')
4.4.6 全局使用:
settings.py:
REST_FRAMEWORK = {
''DEFAULT_PARSER_CLASSES'':[
''rest_framework.parsers.JSONParser''
''rest_framework.parsers.FormParser''
''rest_framework.parsers.MultiPartParser''
]
}
urls.py:
from django.conf.urls import url, include
from web.views import TestView
urlpatterns = [
url(r''test/'', TestView.as_view(), name=''test''),
]
views.py:
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请求,响应内容'')
4.5 序列化(序列化用于对用户请求数据进行验证和数据进行序列化。)
4.5.1 自定义字段:
urls.py:
from django.conf.urls import url, include
from web.views.s6_serializers import TestView
urlpatterns = [
url(r''test/'', TestView.as_view(), name=''test''),
]
views.py:
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请求,响应内容'')
4.5.2 基于Model自动生成字段:
urls.py:
from django.conf.urls import url, include
from web.views.s6_serializers import TestView
urlpatterns = [
url(r''test/'', TestView.as_view(), name=''test''),
]
views.py:
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请求,响应内容'')
4.5.3 生成URL:
urls.py:
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''),
]
views.py:
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请求,响应内容'')
4.5.4 自动生成URL:
urls.py:
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''),
]
views.py:
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请求,响应内容'')
4.6 分页:
4.6.1 根据页码进行分页:
urls.py:
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()),
]
views.py:
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
4.6.2 位置和个数进行分页:
urls.py:
from django.conf.urls import url, include
from web.views import s9_pagination
urlpatterns = [
url(r''^test/'', s9_pagination.UserViewSet.as_view()),
]
views.py:
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
4.6.3 游标分页:
urls.py:
from django.conf.urls import url, include
from web.views import s9_pagination
urlpatterns = [
url(r''^test/'', s9_pagination.UserViewSet.as_view()),
]
views.py:
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
4.7 路由系统:
4.7.1 自定义路由:
urls.py:
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())
]
views.py:
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(''...'')
4.7.2 半自动路由:
urls.py:
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''})),
]
views.py:
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
4.7.3 全自动路由:
urls.py:
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)),
]
views.py:
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
4.8 视图:
4.8.1 GenericViewSet:
urls.py:
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''),
]
views.py:
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
4.8.2 ModelViewSet(自定义URL):
urls.py:
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''})),
]
views.py:
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
4.8.3 ModelViewSet(rest framework路由):
urls.py:
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)),
]
views.py:
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
4.9 渲染器:
4.9.1 简介:
根据 用户请求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
4.9.2 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/
urls.py:
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()),
]
views.py:
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)
4.9.3 表格:
访问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/
views.py:
#!/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)
4.9.4 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/
views.py:
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)
4.9.5 自定义显示模板:
访问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/
urls.py:
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()),
]
views.py:
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'')
userdetail.html:
<!DOCTYPE html>
<html lang = "en">
<head>
<meta charset = "UTF-8">
<title> Title </title>
</head>
<body>
{{user}}
{{pwd}}
{{ut}}
</body>
</html>
4.9.5 浏览器格式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/
views.py:
#!/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 drf:幂等性和幂等性 redis的分享已经结束,谢谢您的关注,如果想了解更多关于2020-05-12:哪些场景需要幂等性?怎么保证幂等性?、Django -- DRF 认证流程、Django 2.0 学习(22):Django CSRF、django DRF的相关知识,请在本站进行查询。
本文标签: