django+drf开发一些个人的标准化

最近在改造一下个人的开发风格。特分享一下。

• 子应用我一般放在apps中，每个不同模块的子应用起不同的名字。startapp后自己移动一下，记得修改一下Appconfig中的name即可。
• 子应用中创建services.py或者如有需要可以创建services模块再细分。所有业务放到services中编写。
• views一律改成apis.py，将views中的业务分割services.py中。
• 序列化器中一般不写业务相关代码。
• 提供一个ApiResponse做统一返回，自己继承DRF的response改写一下即可。配合自己异常类，以及全局的异常枚举状态信息。
• 另外也要改写全局异常捕获处理。做异常统一格式输出。

这里举例一个子应用User。

# apps.users.drf.apis.py
from django.contrib.auth import logout
from rest_framework import serializers
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.request import Request
from rest_framework.views import APIView

from apps.users.drf.services import (
    user_get_login_data,
    user_login,
    user_mobile_exist,
    user_register,
    user_username_exist
)
from utils.drf import ApiResponse
from utils.exception.statuscode import GlobalStatusCode


# 用户注册API
class RegisterApi(APIView):
    permission_classes = (AllowAny,)

    # 将序列化器单独定义在api内中，尽量减少重用序列化器
    class RegisterInputSerializer(serializers.Serializer):
        username = serializers.CharField(required=True)
        password = serializers.CharField(required=True)
        password2 = serializers.CharField(required=True)
        mobile = serializers.CharField(required=True)
        allow = serializers.CharField(required=True)
        sms_code = serializers.CharField(required=True, min_length=6, max_length=6)

    def post(self, request: Request):
        serializer = self.RegisterInputSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        
        # 调用服务层的方法
        user = user_register(request, serializer.validated_data)

        return ApiResponse(GlobalStatusCode.OK if user else GlobalStatusCode.REGISTER_FAILED_ERR)


# 用户登录api
class LoginApi(APIView):
    permission_classes = (AllowAny,)

    class UserLoginSerializer(serializers.Serializer):
        username = serializers.CharField()
        password = serializers.CharField()

    def post(self, request: Request):
        serializer = self.UserLoginSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        
        user = user_login(request, serializer.validated_data)
        data = user_get_login_data(user=user)
        
        return ApiResponse(GlobalStatusCode.OK, data=data)


# 用户登出api
class LogoutApi(APIView):
    permission_classes = (IsAuthenticated,)

    def post(self, request: Request):
        # 这种就没必要再services.py中定义函数来调用业务了，本身就一行简单的业务代码。
        logout(request)
        
        return ApiResponse(GlobalStatusCode.OK)

服务层services.py

# apps/user/drf/services.py
import re

from django.contrib.auth import authenticate, login
from django.db.models import Q
from django_redis import get_redis_connection
from redis import Redis

from apps.users.models import User
from utils.exception.drf import BusinessException
from utils.exception.statuscode import GlobalStatusCode
from utils.regexstring import PHONENUM,PASSWORD,USERNAME


def user_username_exist(username):
    """判断用户名是否存在"""
    return User.objects.filter(username=username, is_superuser=False).count() > 1


def user_mobile_exist(mobile):
    """判断手机号是否存在"""
    return User.objects.filter(mobile=mobile, is_superuser=False).count() > 1


def user_register(request, validated_data) -> User:
    """用户注册业务逻辑"""

    username = validated_data['username']
    mobile = validated_data['mobile']
    sms_code = validated_data['sms_code']

    # 先校验是否勾选协议，因为不勾选全是白瞎，没必要浪费性能区校验这个那个的..
    if validated_data['allow'] != '1':
        raise BusinessException(GlobalStatusCode.ALLOW_ERR)

    # 判断两次输入的密码是否一致
    if validated_data['password'] != validated_data['password2']:
        raise BusinessException(GlobalStatusCode.CPWD_ERR)

    # 只需要校验密码是否符合规定就可以了，确认密码不要校验规则，只需要和密码判断是否相同。
    if re.match(PASSWORD, validated_data['password']) is None:
        raise BusinessException(detail='密码格式不符合规定')

    if re.match(USERNAME, username) is None:
        raise BusinessException(GlobalStatusCode.USER_ERR)

    if re.match(PHONENUM, mobile) is None:
        raise BusinessException(GlobalStatusCode.MOBILE_ERR)

    # 判断用户是否已注册
    user_count = User.objects.filter(Q(username=username) | Q(mobile=mobile)).count()
    if user_count > 0:
        raise BusinessException(detail='用户已经注册')

    # 校验短信验证码
    from django.conf import settings
    if settings.DEBUG:
        pass
        # 测试环境直接跳过校验验证码
        print('测试跳过验证码..')
    else:
        redis_connection: Redis = get_redis_connection()
        code_from_redis = redis_connection.get(validated_data['mobile'])
        if code_from_redis is None:
            raise BusinessException(detail='短信验证码已过期！请重新获取！')

        if code_from_redis.decode() != sms_code:
            raise BusinessException(detail='短信验证码不正确！请重新输入！')

    user = User.objects.create_user(
        username=validated_data['username'],
        password=validated_data['password'],
        mobile=validated_data['mobile']
    )

    # 登录
    login(request, user)

    return user


def user_get_login_data(*, user: User):
    return {
        "id": user.id,
        "mobile": user.mobile,
        "username": user.username,
        "email": user.email,
        "is_superuser": user.is_superuser,
    }


def user_login(request, validated_data):
    username = validated_data['username']

    if re.match(PHONENUM, username) is not None:
        User.USERNAME_FIELD = "mobile"

    user = authenticate(**validated_data)

    if user is None:
        raise BusinessException(GlobalStatusCode.PWD_ERR)

    login(request, user)

    return user