实战-DRF快速写接口

开发环境

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

测试工具

Postman

需求

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

模型

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

序列化器

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__'

视图

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

认证权限频率

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

配置文件

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

]

测试

下面是普通用户，403了~

巴特西

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

实战-DRF快速写接口

开发环境

测试工具

需求

模型

序列化器

视图

认证权限频率

配置文件

路由

测试

最新文章

热门文章