models.py
from django.db import models class UserInfo(models.Model): name=models.CharField(max_length=32) pwd=models.CharField(max_length=32) type_choices=((1,"普通用户"),(2,"VIP"),(3,"SVIP")) user_type=models.IntegerField(choices=type_choices,default=1) class Token(models.Model): user=models.OneToOneField("UserInfo") token = models.CharField(max_length=128) def __str__(self): return self.token class Book(models.Model): title=models.CharField(max_length=32) price=models.IntegerField() pub_date=models.DateField() publish=models.ForeignKey("Publisher") authors=models.ManyToManyField("Author") def __str__(self): return self.title class Publisher(models.Model): name=models.CharField(max_length=32) email=models.EmailField() def __str__(self): return self.name class Author(models.Model): name=models.CharField(max_length=32) age=models.IntegerField() def __str__(self): return self.name
views.py
from django.shortcuts import render from app01 import models from rest_framework import serializers from rest_framework.response import Response from rest_framework.views import APIView class PublisherModelSerializer(serializers.ModelSerializer): class Meta: model = models.Publisher fields = "__all__" class BookModelSerializer(serializers.ModelSerializer): class Meta: model = models.Book fields = "__all__" class AuthorModelSerializer(serializers.ModelSerializer): class Meta: model = models.Author fields = "__all__" #****用户登录认证组件 from rest_framework.authentication import BaseAuthentication from rest_framework.exceptions import AuthenticationFailed class AuthToken(BaseAuthentication): def authenticate(self, request): token = request.query_params.get('token') #query_params函数返回的是self._request.GET obj = models.Token.objects.filter(token=token).first() if not obj: raise AuthenticationFailed({'code':1001,'error':'认证失败'}) return (obj.user.name,obj) #*********权限组件**** from rest_framework.permissions import BasePermission class SVIPPermission(BasePermission): #继承于BasePermission，/authous/2/才能查看每个具体的作者。否则报错 message = "只有超级用户才能访问" def has_permission(self,request,view): username = request.user print(username) user_type = models.UserInfo.objects.filter(name=username).first().user_type if user_type == 3: return True else: return False ##*******频率组件 # from rest_framework.throttling import BaseThrottle # VISIT_RECORD={} # class VisitThrottle(BaseThrottle): # # def __init__(self): # self.history=None # # def allow_request(self,request,view): # remote_addr = request.META.get('REMOTE_ADDR') # print(remote_addr) # import time # ctime=time.time() # # if remote_addr not in VISIT_RECORD: # VISIT_RECORD[remote_addr]=[ctime,] # return True # # history=VISIT_RECORD.get(remote_addr) # self.history=history # # while history and history[-1]<ctime-60: # history.pop() # # if len(history)<10: # history.insert(0,ctime) # return True # else: # return False # # def wait(self): # import time # ctime=time.time() # return 60-(ctime-self.history[-1]) class PublisherView(APIView): #获取所有出版社 def get(self,request): pub_list = models.Publisher.objects.all() pub_ser = PublisherModelSerializer(pub_list,many=True) return Response(pub_ser.data) #提交出版社 def post(self,request): pub_ser = PublisherModelSerializer(data=request.data) if pub_ser.is_valid(): print(pub_ser.validated_data) pub_ser.save() return Response(pub_ser.data) else: return Response(pub_ser.errors) class PublisherDetailView(APIView): #查看某个出版社 def get(self,request,pk): pub_obj = models.Publisher.objects.filter(pk=pk).first() pub_ser = PublisherModelSerializer(pub_obj) return Response(pub_ser.data) #更新某个出版社 def put(self,request,pk): pub_obj = models.Publisher.objects.filter(pk=pk).first() pub_ser = PublisherModelSerializer(pub_obj,data=request.data) if pub_ser.is_valid(): pub_ser.save() return Response(pub_ser.data) else: return Response(pub_ser.errors) #删除某个出版社 def delete(self,request,pk): models.Publisher.objects.filter(pk=pk).delete() return Response() #****解析器
###--->根据contenttype的不同解析成不同找到不同的解析器，解析出的是字典或者QueryDict ##无论post(请求)前面传的是JSON数据还是urlencoded数据，rest framework 中的request.data都 # 帮我们解析成字典 而django中的request.POST只解析urlencode数据 from rest_framework.parsers import JSONParser,FormParser,MultiPartParser,FileUploadParser #默认使用前三个解析器 # 在视图类中添加 parser_classes = [JSONPaser,] #如果只写这一个，意味着只能解析JSON数据 FormParser解析urlencode数据 #*********渲染器***** from rest_framework.renderers import JSONRenderer,BrowsableAPIRenderer #在视图类中添加 renderer_classes = [JSONRenderer,] #****响应器就是Response比django原生的HttpResponse多了一些功能 #*******版本*** from rest_framework.versioning import QueryParameterVersioning,URLPathVersioning # versioning_class = URLPathVersioning request.version,获取版本的值推荐用URLPathVersioning不用QueryParameterVersioning # # ******第二种分页 ********* from rest_framework.pagination import PageNumberPagination class MyPageNumberPagination(PageNumberPagination): page_size = 2 # 每页显示多少条数据 page_query_param = "page" # URL中输入的 page=2 page_size_query_param = "size" # 临时浏览器中使用127.0.0.1:8000/books/?page=2&size=2 第二页显示两条数据 max_page_size = 3 # 上面的size最大值，不能无限制显示10000条数据 # # ******第三种分页 ********* from rest_framework.pagination import PageNumberPagination,LimitOffsetPagination class MyLimitOffsetPagination(LimitOffsetPagination): default_limit = 1 #每页显示多少条数据 class BookView(APIView): # authentication_classes = [AuthToken,] # permission_classes = [SVIPPermission,] #依赖于上面的AuthToken # throttle_classes = [] #频率 # parser_classes = [JSONParser, FormParser] #解析器 # renderer_classes = [JSONRenderer,] #渲染器JSONRenderer只返回json数据，不带html页面 # versioning_class = QueryParameterVersioning # def get(self,request): # # book_list = models.Book.objects.all() # # from rest_framework.pagination import PageNumberPagination # # ******第一种分页 不常用********* # #在settings中配置"PAGE_SIZE"的值,定义多少条数据就显示多少 不常用 # pnp = PageNumberPagination() # book_pag = pnp.paginate_queryset(book_list,request,self) # book_ser = BookModelSerializer(book_pag,many=True) # return Response(book_ser.data) def get(self,request,*args,**kwargs): book_list = models.Book.objects.all() #**第二种分页 pnp = MyPageNumberPagination() book_pag = pnp.paginate_queryset(book_list,request,self) book_ser = BookModelSerializer(book_pag,many=True) return Response(book_ser.data) # def get(self,request): # # book_list = models.Book.objects.all() # #**第三种分页 # pnp = MyLimitOffsetPagination() # book_pag = pnp.paginate_queryset(book_list,request,self) # book_ser = BookModelSerializer(book_pag,many=True) # # return Response(book_ser.data) def post(self,request,*args,**kwargs): book_ser = BookModelSerializer(data=request.data) if book_ser.is_valid(): print(book_ser.validated_data) book_ser.save() return Response(book_ser.data) else: return Response(book_ser.errors) class BookDetailView(APIView): def get(self,request,pk,*args,**kwargs): book_obj = models.Book.objects.filter(pk=pk).first() book_ser = BookModelSerializer(book_obj) return Response(book_ser.data) def put(self,request,pk,*args,**kwargs): book_obj = models.Book.objects.filter(pk=pk).first() book_ser = BookModelSerializer(book_obj,data=request.data) if book_ser.is_valid(): book_ser.save() return Response(book_ser.data) else: return Response(book_ser.errors) def delete(self,request,pk,*args,**kwargs): models.Book.objects.filter(pk=pk).delete() return Response() ##***局部的频率组件 from app01.utils import VisitThrottle from rest_framework import viewsets class BookViewSet(viewsets.ModelViewSet): # authentication_classes = [AuthToken, ] # permission_classes=[SVIPPermission,] throttle_classes = [VisitThrottle] # 限制某个IP每分钟访问次数不能超过20次 queryset = models.Book.objects.all() serializer_class = BookModelSerializer #ModelViewSet中的分页 pagination_class = MyPageNumberPagination import uuid class AuthView(APIView): #authentication_classes = [] # 全局配置后，不让login验证，直接加空列表就不验证 def post(self,request,*args,**kwargs): ##用户登录认证 ret = {'code':1000} user = request.data.get('user') pwd = request.data.get('pwd') user = models.UserInfo.objects.filter(user=user,pwd=pwd).first() if not user: ret['code'] = 1001 ret['error'] = '用户名或密码错误' else: uid = str(uuid.uuid4()) models.Token.objects.update_or_create(user=user,defaults={'token':uid}) ret['token'] = uid return Response(ret)
utils.py 认证，权限，频率
from rest_framework.authentication import BaseAuthentication from rest_framework.exceptions import AuthenticationFailed from app01 import models #*********用户登录认证组件**** class AuthToken(BaseAuthentication): def authenticate(self, request): token = request.query_params.get('token') obj = models.Token.objects.filter(token=token).first() if not obj: raise AuthenticationFailed({'code':1001,'error':'认证失败'}) return (obj.user.name,obj) #*********权限组件**** from rest_framework.permissions import BasePermission class SVIPPermission(BasePermission): #继承于BasePermission，/authous/2/才能查看每个具体的作者。否则报错 message = "只有超级用户才能访问" def has_permission(self,request,view): username = request.user print(username) user_type = models.UserInfo.objects.filter(name=username).first().user_type if user_type == 3: return True else: return False #*********频率组件**** from rest_framework.throttling import BaseThrottle VISIT_RECORD={} class VisitThrottle(BaseThrottle): def __init__(self): self.history=None def allow_request(self,request,view): remote_addr = request.META.get('REMOTE_ADDR') print(remote_addr) import time ctime=time.time() if remote_addr not in VISIT_RECORD: VISIT_RECORD[remote_addr]=[ctime,] return True history=VISIT_RECORD.get(remote_addr) self.history=history while history and history[-1]<ctime-60: history.pop() if len(history)<10: history.insert(0,ctime) return True else: return False def wait(self): import time ctime=time.time() return 60-(ctime-self.history[-1])
setting.py
REST_FRAMEWORK = { # "DEFAULT_AUTHENTICATION_CLASSES": ["app01.utils.TokenAuth",], # "DEFAULT_PERMISSION_CLASSES": ["app01.utils.SVIPPermission",], # "DEFAULT_THROTTLE_CLASSES":["app01.utils.VisitThrottle",], #'DEFAULT_RENDERER_CLASSES':['rest_framework.renderers.JSONRenderer','rest_framework.renderers.BrowsableAPIRenderer',], # "PAGE_SIZE":2, 'DEFAULT_VERSIONING_CLASS':'rest_framework.versioning.URLPathVersioning', 'ALLOWED_VERSIONS':['v1','v2'], # 允许的版本 'VERSION_PARAM':'version', # 参数 'DEFAULT_VERSION':'v1', # 默认版本 } #如果加版本url要如下配置(?P<version>\w+)，如果继承的是APIview,方法中要加*args,**kwargs， # 如果不加版本就不用加*args,**kwargs，获取版本用request.version, #url(r'^(?P<version>\w+)/book/$', views.BookView.as_view()),request.version,获取版本 #url(r'^(?P<version>[v1|v2]+)/users/$', users_list, name='users-list'),这种只能写v1/v2，所以用w+
urls.py
from django.conf.urls import url from django.contrib import admin from app01 import views urlpatterns = [ url(r'^admin/', admin.site.urls), url(r'^publisher/$', views.PublisherView.as_view()), url(r'^publisher/(\d+)/$', views.PublisherDetailView.as_view()), # url(r'^book/$', views.BookView.as_view()), # url(r'^book/(?P<pk>\d+)/$', views.BookDetailView.as_view()), #加版本 url(r'^(?P<version>\w+)/book/$', views.BookView.as_view()), url(r'^(?P<version>\w+)/book/(?P<pk>\d+)/$', views.BookDetailView.as_view()), url(r'^books/$', views.BookViewSet.as_view({"get":"list","post":"create"}),name="book_list"), url(r'^books/(?P<pk>\d+)/$', views.BookViewSet.as_view({ 'get': 'retrieve', 'put': 'update', 'patch': 'partial_update', 'delete': 'destroy' }),name="book_detail"), url(r'^auth/$', views.AuthView.as_view()), ]
本站文章如无特殊说明，均为本站原创，如若转载，请注明出处：django rest_framework中的APIView,ModelViewSet,认证，权限，频率，版本 - Python技术站
微信扫一扫 
支付宝扫一扫 
