实战-DRF快速写接口(认证权限频率) 实战-DRF快速写接口(认证权限频

实战-DRF快速写接口开发环境

Python3.6
Pycharm专业版2021.2.3
Sqlite3
Django 2.2
djangorestframework3.13

测试工具 Postman

文章图片

需求

注册接口，包含字段用户名，密码，确认密码，用户类型
登陆接口，校验用户名，密码，生成随机字符串
认证功能，除了注册登陆接口外，所有接口都要登陆后访问
频率限制功能，每分钟访问5次，book的所有接口，使用这个频率类
权限限制功能，publish的所有操作需要超级用户能访问，其他的普通登陆用户就可以操作

模型

from django.db import modelsclass User(models.Model): username = models.CharField(max_length=32) password = models.CharField(max_length=255) user_type = models.IntegerField(choices=((1, '超级管理员'), (2, '普通管理员'), (3, '普通用户')))class UserToken(models.Model): user = models.OneToOneField(to=User,on_delete=models.CASCADE) token = models.CharField(max_length=32)class Book(models.Model): title = models.CharField(max_length=11) price = models.DecimalField(max_digits=5, decimal_places=2) authors = models.ManyToManyField(to='Author') publish = models.ForeignKey(to='Publish', on_delete=models.CASCADE)@property def publish_detail(self): return {'name': self.publish.name,'Email':self.publish.email}@property def author_list(self): l = [] print(self.authors.all())# for author in self.authors.all(): print(author.authordetail)# AuthorDetail object (1) l.append({'name': author.username, 'gender': author.gender, 'address': author.authordetail.address,'telephone':author.authordetail.telephone}) return lclass Author(models.Model): username = models.CharField(max_length=11) gender = models.IntegerField(choices=((1, '男'), (2, '女'), (3, '未知'))) authordetail = models.OneToOneField(to='AuthorDetail', on_delete=models.CASCADE)@property def authordetail_info(self): return {'telephone': self.authordetail.telephone, 'address': self.authordetail.address}class AuthorDetail(models.Model): telephone = models.BigIntegerField() address = models.CharField(max_length=32)class Publish(models.Model): name = models.CharField(max_length=11) email = models.EmailField()

序列化器

from rest_framework import serializersfrom .models import *# 用户序列化器 class UserSerializer(serializers.ModelSerializer): class Meta: model = User fields = '__all__'# 书序列化器 class BookSerializer(serializers.ModelSerializer): class Meta: model = Book # fields = '__all__' fields = ['id', 'title', 'price', 'publish', 'authors', 'publish_detail', 'author_list']extra_kwargs = { 'publish': {'write_only': True}, 'authors': {'write_only': True}, }# 作者序列化器 class AuthorSerializer(serializers.ModelSerializer): class Meta: # 指定和哪个表有关系 model = Author # fields = '__all__' fields = ['id', 'username', 'gender', 'telephone', 'address', 'authordetail_info']# 重写字段telephone和addr telephone = serializers.CharField(write_only=True) address = serializers.CharField(write_only=True, max_length=8, required=False)# 重写create，操作两个表 def create(self, validated_data): # 先存作者详情 authordetail = AuthorDetail.objects.create(telephone=validated_data.get('telephone'), address=validated_data.get('address')) # 存作者表 author = Author.objects.create(author_detail=authordetail, gender=validated_data.get('gender'), username=validated_data.get('username')) # 这样只返回author对象就行，直接存了两个表,返回反序列化的对象 return author# 出版社序列化器 class PublishSerializer(serializers.ModelSerializer): class Meta(): model = Publish fields = '__all__'

视图

from django.contrib.auth.hashers import make_password, check_password from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.viewsets import ViewSet from rest_framework.viewsets import ModelViewSet from .auth import * from .serializer import * from .models import *# 注册视图 class UserRegisterView(ViewSet): @action(methods=["POST"], detail=False) def register(self, request): usernmae = request.data.get('username') password = request.data.get('password') re_password = request.data.get('re_password') user_type = request.data.get('user_type') if User.objects.filter(username=usernmae): return Response({'msg': f'用户{usernmae}已注册！', 'code': 4000}) else: if password == re_password: # make_password加密:make_password(password, salt=None, hasher='default') user_date = {'username': usernmae, 'password': make_password(password), 'user_type': user_type} user_serializer = UserSerializer(data=https://www.it610.com/article/user_date) if user_serializer.is_valid(): user_serializer.save() return Response({'code': 2001, 'msg': f'用户{usernmae}注册成功'}) else: return Response({'code': 4001, 'msg': '注册失败', 'errors': user_serializer.errors}) else: return Response({'msg': '两次密码不一致', 'code': 4002})# 登录视图 class UserLoginView(ViewSet):@action(methods=["POST"], detail=False) def login(self, request): username = request.data.get('username') password = request.data.get('password') user = User.objects.filter(username=username).first() # check_password(password, encoded, setter=None, preferred='default') if user and check_password(password, user.password): import uuid token = str(uuid.uuid4()) UserToken.objects.update_or_create(user=user, defaults={'token': token}) return Response({'code': 2000, 'msg': f'用户{user.username}登录成功', 'token': token}) return Response({'code': 4004, 'msg': '校验失败，用户名或密码错误'})# 书接视图 class BookView(ModelViewSet): authentication_classes = [LoginAuth,] throttle_classes = [IPThrottle,] queryset = Book.objects.all() serializer_class = BookSerializer# 作者视图 class AuthorView(ModelViewSet): authentication_classes = [LoginAuth,] queryset = Author.objects.all() serializer_class = AuthorSerializer# 出版社视图 class PublishView(ModelViewSet): authentication_classes = [LoginAuth, ] permission_classes = [UserPermission,] queryset = Publish.objects.all() serializer_class = PublishSerializer

认证权限频率

from rest_framework.authentication import BaseAuthentication from rest_framework.exceptions import AuthenticationFailed from rest_framework.permissions import BasePermission from rest_framework.throttling import SimpleRateThrottlefrom app01 import models# 认证类 class LoginAuth(BaseAuthentication): # 重写authenticate方法 def authenticate(self, request): # 获取前端携带的token,token放在哪是自己规定的，比如从查询参数中获取 token = request.query_params.get('token') # 比对随机字符串 user_token = models.UserToken.objects.filter(token=token).first() if user_token: # 登录了，返回当前登录用户和token return user_token.user, token else: # 没有登录，抛异常 raise AuthenticationFailed('您没有登录，请登录')# 权限类 class UserPermission(BasePermission): def has_permission(self, request, view): # 没有权限的提示信息 self.message = '您是:%s，没有权限' % request.user.get_user_type_display() # 如果有权限，返回True,没有权限返回False # 权限类，在认证类之后，request.user有了当前登录用户 user_type = request.user.user_type print(user_type) if user_type < 3:# 只要不是1，2，就没有权限 return True else: return False# 频率类 class IPThrottle(SimpleRateThrottle): scope = 'ip'# get_cache_key返回什么就以什么方法做限制，限制条件必须唯一，比如用户id def get_cache_key(self, request, view): # 限制ip地址，从request.META字典中获取ip ''' request.META:请求头中的数据 ''' return request.META.get('REMOTE_ADDR')# 客户端ip

配置文件

REST_FRAMEWORK = { 'DEFAULT_THROTTLE_RATES': { 'ip': '5/m'# minute_3是scope的字符串，一分钟访问5次 }, }

路由

from django.contrib import admin from django.urls import path,include from app01 import views from rest_framework.routers import SimpleRouter router = SimpleRouter() router.register('user',views.UserLoginView,'user') router.register('user',views.UserRegisterView,'user')router.register('books',views.BookView,'books') router.register('author',views.AuthorView,'author') router.register('publish',views.PublishView,'publish')urlpatterns = [ path('admin/', admin.site.urls), path('', include(router.urls)), ]

测试