rest_framework 视图/路由/渲染器/认证授权/节流

视图

1. 整理

class View： django view

class APIView(View): rest框架基础view，适合直接继承，重写get post方法

class GenericAPIView(views.APIView): 对ApiView进一步包装，方法重写，不适合直接继承使用，适合配合mixin

功能扩展，以下各实现部分 get post delete put patch 方法，但未重写asview，需要进一步改写，奇怪的类。。。

ViewSetMixin 重写了as_view方法，可以把httpmethod的 get post delete put patch转化为不同方法。

GenericViewSet 可以自定制，适合直接继承，转化httpmethod的 get post delete put patch转化为自定制方法。也可已结合部分XXModelViewSet

ModelViewSet 高度整合，适合直接继承，用于增删改查可以把httpmethod的 get post delete put patch转化为list，create，destroy，update，pratial_update

总结

2. modelviewset

views.py

class TestViewSet(ModelViewSet):
    queryset = UserInfo.objects.all()
    serializer_class = UserInfoSerializer
    pagination_class=MyPagination

View Code

urls.py

 path('testviewset/',app01views.TestViewSet.as_view({'get':'list','post':'create'})),
    re_path('testviewset/(?P<pk>d+)/',app01views.TestViewSet.as_view({'get':'retrieve','delete':'destroy','put':'update','patch':'partial_update'})),

View Code

由于Response返回形式，加入？format=json返回json或者在url改写

    # http://127.0.0.1:8000/testviewset/json
    re_path('testviewset/(?P<format>w+)',app01views.TestViewSet.as_view({'get':'list','post':'create'})),

View Code

自动路由

当实现了modelviewset时的的自动路由

from rest_framework import routers

router=routers.DefaultRouter()
router.register(r'testrouter',app01views.TestViewSet)

urlpatterns = [
    path('',include(router.urls)),
]

#访问http://127.0.0.1:8000/testrouter/

View Code

或

from rest_framework import routers

router=routers.DefaultRouter()
router.register(r'',app01views.TestViewSet)

urlpatterns = [
    path('testrouter/',include(router.urls)),
]

View Code

渲染器

from rest_framework.renderers import AdminRenderer,JSONRenderer,BrowsableAPIRenderer
class TestViewSet(ModelViewSet):
    renderer_classes = [JSONRenderer,BrowsableAPIRenderer]
    queryset = UserInfo.objects.all()
    serializer_class = UserInfoSerializer
    pagination_class=MyPagination

View Code

认证授权

参考http://www.cnblogs.com/wupeiqi/articles/7805382.html

from rest_framework.authentication import BaseAuthentication
from rest_framework.permissions import BasePermission

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

View Code

全局

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

View Code

节流

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)

View Code