GVKun编程网logo

Django drf:幂等性(幂等性 redis)

25

在这篇文章中,我们将为您详细介绍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)

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:哪些场景需要幂等性?怎么保证幂等性?

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 -- 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 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

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的相关知识,请在本站进行查询。

本文标签: