实战-DRF快速写接口
开发环境
- Python3.6
- Pycharm专业版2021.2.3
- Sqlite3
- Django 2.2
- djangorestframework3.13
测试工具
Postman
需求
- 注册接口，包含字段用户名，密码，确认密码，用户类型
- 登陆接口，校验用户名，密码，生成随机字符串
- 认证功能，除了注册登陆接口外，所有接口都要登陆后访问
- 频率限制功能，每分钟访问5次，book的所有接口，使用这个频率类
- 权限限制功能，publish的所有操作需要超级用户能访问，其他的普通登陆用户就可以操作
模型
代码语言：javascript复制from django.db import models
class 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()) # <QuerySet [<Author: Author object (1)>, <Author: Author object (2)>]>
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 l
class 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()序列化器
代码语言：javascript复制from rest_framework import serializers
from .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__'视图
代码语言：javascript复制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=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认证权限频率
代码语言：javascript复制from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from rest_framework.permissions import BasePermission
from rest_framework.throttling import SimpleRateThrottle
from 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配置文件
代码语言：javascript复制REST_FRAMEWORK = {
'DEFAULT_THROTTLE_RATES': {
'ip': '5/m' # minute_3是scope的字符串，一分钟访问5次
}, }路由
代码语言：javascript复制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)),
]测试
下面是普通用户，403了~
