Django Rest Framework

Django Rest Framework

一、什么是RESTful

  • REST与技术无关，代表的是一种软件架构风格，REST是Representational State Transfer的简称，中文翻译为“表征状态转移”

  • REST从资源的角度类审视整个网络，它将分布在网络中某个节点的资源通过URL进行标识，客户端应用通过URL来获取资源的表征，获得这些表征致使这些应用转变状态

  • 所有的数据，不过是通过网络获取的还是操作（增删改查）的数据，都是资源，将一切数据视为资源是REST区别与其他架构风格的最本质属性

  • 对于REST这种面向资源的架构风格，有人提出一种全新的结构理念，即：面向资源架构（ROA：Resource Oriented Architecture）

二、RESTful API设计

API与用户的通信协议，总是使用HTTPs协议。

域名

版本

**路径，视网络上任何东西都是资源，均使用名词表示（可复数） **

method

  • GET ：从服务器取出资源（一项或多项）
  • POST ：在服务器新建一个资源
  • PUT ：在服务器更新资源（客户端提供改变后的完整资源）
  • PATCH ：在服务器更新资源（客户端提供改变的属性）
  • DELETE ：从服务器删除资源

过滤，通过在url上传参的形式传递搜索条件

状态码

200 OK - [GET]：服务器成功返回用户请求的数据，该操作是幂等的（Idempotent）。
201 CREATED - [POST/PUT/PATCH]：用户新建或修改数据成功。
202 Accepted - [*]：表示一个请求已经进入后台排队（异步任务）
204 NO CONTENT - [DELETE]：用户删除数据成功。
400 INVALID REQUEST - [POST/PUT/PATCH]：用户发出的请求有错误，服务器没有进行新建或修改数据的操作，该操作是幂等的。
401 Unauthorized - [*]：表示用户没有权限（令牌、用户名、密码错误）。
403 Forbidden - [*] 表示用户得到授权（与401错误相对），但是访问是被禁止的。
404 NOT FOUND - [*]：用户发出的请求针对的是不存在的记录，服务器没有进行操作，该操作是幂等的。
406 Not Acceptable - [GET]：用户请求的格式不可得（比如用户请求JSON格式，但是只有XML格式）。
410 Gone -[GET]：用户请求的资源被永久删除，且不会再得到的。
422 Unprocesable entity - [POST/PUT/PATCH] 当创建一个对象时，发生一个验证错误。
500 INTERNAL SERVER ERROR - [*]：服务器发生错误，用户将无法判断发出的请求是否成功。

更多看这里：http://www.w3.org/Protocols/rfc2616/rfc2616-sec10.html

错误处理，状态码是4xx时，应返回错误信息，error当做key。

{
    error: "Invalid API key"
}

返回结果，针对不同操作，服务器向用户返回的结果应该符合以下规范。

GET /collection：返回资源对象的列表（数组）
GET /collection/resource：返回单个资源对象
POST /collection：返回新生成的资源对象
PUT /collection/resource：返回完整的资源对象
PATCH /collection/resource：返回完整的资源对象
DELETE /collection/resource：返回一个空文档

Hypermedia API，RESTful API最好做到Hypermedia，即返回结果中提供链接，连向其他API方法，使得用户不查文档，也知道下一步应该做什么。

{"link": {
  "rel":   "collection https://www.example.com/zoos",
  "href":  "https://api.example.com/zoos",
  "title": "List of zoos",
  "type":  "application/vnd.yourformat+json"
}}

摘自：http://www.ruanyifeng.com/blog/2014/05/restful_api.html

三、基于Django实现

路由系统：

urlpatterns = [
    url(r'^users', Users.as_view()),
]

CBV视图：

from django.views import View
from django.http import JsonResponse
 
class Users(View):
    def get(self, request, *args, **kwargs):
        result = {
            'status': True,
            'data': 'response data'
        }
        return JsonResponse(result, status=200)
 
    def post(self, request, *args, **kwargs):
        result = {
            'status': True,
            'data': 'response data'
        }
        return JsonResponse(result, status=200)

四、基于Django Rest Framework框架实现

1、基本流程

url.py

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

views.py

from rest_framework.views import APIView
from rest_framework.response import Response
 
 
class TestView(APIView):
    def dispatch(self, request, *args, **kwargs):
        """
        请求到来之后，都要执行dispatch方法，dispatch方法根据请求方式不同触发 get/post/put等方法
         
        注意：APIView中的dispatch方法有好多好多的功能
        """
        return super().dispatch(request, *args, **kwargs)
 
    def get(self, request, *args, **kwargs):
        return Response('GET请求，响应内容')
 
    def post(self, request, *args, **kwargs):
        return Response('POST请求，响应内容')
 
    def put(self, request, *args, **kwargs):
        return Response('PUT请求，响应内容')

上述是rest framework框架基本流程，重要的功能是在APIView的dispatch中触发。

2.认证和授权

a. 用户url传入的token认证

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

b. 请求头认证

  • urls.py

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

c.多个认证规则

  • 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

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

d.认证和权限

  • 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.authentication import BaseAuthentication
  from rest_framework.permissions import BasePermission
  
  from rest_framework.request import Request
  from rest_framework import exceptions
  
  token_list = [
      'sfsfss123kuf3j123',
      'asijnfowerkkf9812',
  ]
  
  
  class TestAuthentication(BaseAuthentication):
      def authenticate(self, request):
          """
          用户认证，如果验证成功后返回元组： (用户,用户Token)
          :param request: 
          :return: 
              None,表示跳过该验证；
                  如果跳过了所有认证，默认用户和Token和使用配置文件进行设置
                  self._authenticator = None
                  if api_settings.UNAUTHENTICATED_USER:
                      self.user = api_settings.UNAUTHENTICATED_USER() # 默认值为：匿名用户
                  else:
                      self.user = None
          
                  if api_settings.UNAUTHENTICATED_TOKEN:
                      self.auth = api_settings.UNAUTHENTICATED_TOKEN()# 默认值为：None
                  else:
                      self.auth = None
              (user,token)表示验证通过并设置用户名和Token；
              AuthenticationFailed异常
          """
          val = request.query_params.get('token')
          if val not in token_list:
              raise exceptions.AuthenticationFailed("用户认证失败")
  
          return ('登录用户', '用户token')
  
      def authenticate_header(self, request):
          """
          Return a string to be used as the value of the `WWW-Authenticate`
          header in a `401 Unauthenticated` response, or `None` if the
          authentication scheme should return `403 Permission Denied` responses.
          """
          pass
  
  
  class TestPermission(BasePermission):
      message = "权限验证失败"
  
      def has_permission(self, request, view):
          """
          判断是否有权限访问当前请求
          Return `True` if permission is granted, `False` otherwise.
          :param request: 
          :param view: 
          :return: True有权限；False无权限
          """
          if request.user == "管理员":
              return True
  
      # GenericAPIView中get_object时调用
      def has_object_permission(self, request, view, obj):
          """
          视图继承GenericAPIView，并在其中使用get_object时获取对象时，触发单独对象权限验证
          Return `True` if permission is granted, `False` otherwise.
          :param request: 
          :param view: 
          :param obj: 
          :return: True有权限；False无权限
          """
          if request.user == "管理员":
              return True
  
  
  class TestView(APIView):
      # 认证的动作是由request.user触发
      authentication_classes = [TestAuthentication, ]
  
      # 权限
      # 循环执行所有的权限
      permission_classes = [TestPermission, ]
  
      def get(self, request, *args, **kwargs):
          # self.dispatch
          print(request.user)
          print(request.auth)
          return Response('GET请求，响应内容')
  
      def post(self, request, *args, **kwargs):
          return Response('POST请求，响应内容')
  
      def put(self, request, *args, **kwargs):
          return Response('PUT请求，响应内容')

e. 全局使用

上述操作中均是对单独视图进行特殊配置，如果想要对全局进行配置，则需要再配置文件中写入即可

  • 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

      #!/usr/bin/env python
  # -*- coding:utf-8 -*-
  from rest_framework.views import APIView
  from rest_framework.response import Response
  
  class TestView(APIView):
  
      def get(self, request, *args, **kwargs):
          # self.dispatch
          print(request.user)
          print(request.auth)
          return Response('GET请求，响应内容')
  
      def post(self, request, *args, **kwargs):
          return Response('POST请求，响应内容')
  
      def put(self, request, *args, **kwargs):
          return Response('PUT请求，响应内容')

3. 用户访问次数/频率限制

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

b. 基于用户IP显示访问频率（利于Django缓存）

  • 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

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

c. view中限制请求频率

  • 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):
          """
          Should return a unique cache-key which can be used for throttling.
          Must be overridden.
  
          May return `None` if the request should not be throttled.
          """
          if not request.user:
              ident = self.get_ident(request)
          else:
              ident = request.user
  
          return self.cache_format % {
              'scope': self.scope,
              'ident': ident
          }
  
  
  class TestView(APIView):
      throttle_classes = [TestThrottle, ]
  
      # 在settings中获取 xxxxxx 对应的频率限制值
      throttle_scope = "xxxxxx"
  
      def get(self, request, *args, **kwargs):
          # self.dispatch
          print(request.user)
          print(request.auth)
          return Response('GET请求，响应内容')
  
      def post(self, request, *args, **kwargs):
          return Response('POST请求，响应内容')
  
      def put(self, request, *args, **kwargs):
          return Response('PUT请求，响应内容')
  
      def throttled(self, request, wait):
          """
          访问次数被限制时，定制错误信息
          """
  
          class Throttled(exceptions.Throttled):
              default_detail = '请求被限制.'
              extra_detail_singular = '请 {wait} 秒之后再重试.'
              extra_detail_plural = '请 {wait} 秒之后再重试.'
  
          raise Throttled(wait)

d. 匿名时用IP限制+登录时用Token限制

  • 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

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

e. 全局使用

  • settings

      REST_FRAMEWORK = {
      'DEFAULT_THROTTLE_CLASSES': [
          'api.utils.throttles.throttles.LuffyAnonRateThrottle',
          'api.utils.throttles.throttles.LuffyUserRateThrottle',
      ],
      'DEFAULT_THROTTLE_RATES': {
          'anon': '10/day',
          'user': '10/day',
          'luffy_anon': '10/m',
          'luffy_user': '20/m',
      },
  }

4.版本

a. 基于url的get传参方式

如：/users?version=v1

  • settings

      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

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

b. 基于url的正则方式

如：/v1/users/

  • 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

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

c. 基于 accept 请求头方式

如：Accept: application/json; version=1.0

  • 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

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

d. 基于主机名方法

如：v1.example.com

  • 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

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

e. 基于django路由系统的namespace

如：example.com/v1/users/

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

  • 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

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

f. 全局使用

  • settings.py

      REST_FRAMEWORK = {
      'DEFAULT_VERSIONING_CLASS':"rest_framework.versioning.URLPathVersioning",
      'DEFAULT_VERSION': 'v1',
      'ALLOWED_VERSIONS': ['v1', 'v2'],
      'VERSION_PARAM': 'version' 
  }

5.解析器（parser）

根据请求头 content-type 选择对应的解析器就请求体内容进行处理。

a. 仅处理请求头content-type为application/json的请求体

  • 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

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

b. 仅处理请求头content-type为application/x-www-form-urlencoded 的请求体

  • 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

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

c. 仅处理请求头content-type为multipart/form-data的请求体

  • 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

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

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

d. 仅上传文件

  • 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

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

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

e. 同时多个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

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

f. 全局使用

  • 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

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

注意：个别特殊的值可以通过Django的request对象 request._request 来进行获取

6.序列化

序列化用于对用户请求数据进行验证和数据进行序列化。

a. 自定义字段

  • 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

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

b. 基于Model自动生成字段

  • 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

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

c. 生成URL

  • 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

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