最近在改造一下个人的开发风格。特分享一下。
- 子应用我一般放在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