牛年第一篇文章，金鼠辞旧岁、金牛报春时，我携全家以及秘书安南、保镖普京、管家布莱尔、司机布什、家奴小泉、水扁等向各位读者大佬朋友们致以节日的问候：祝新年快乐，玩得开心o(￣▽￣)o 如需查看本项目实际运行效果，可点击uni-app实战之社区交友APP（1）项目介绍和环境搭建（免费试读）进行浏览。 如需本项目完整前端uni-app代码和资源文件，可以点击https://download.csdn.net/download/CUFEECR/15316002下载，或者订阅本专栏uni-app社区交友APP开发实战，即可在本专栏下序号为5的整数倍的博客（例如uni-app实战之社区交友APP（5）搜索和发布页开发和uni-app实战之社区交友APP（10）登录、个人空间开发和动画优化）文末获取百度网盘链接和提取码。同时为了感谢各位读者的支持，订阅本专栏的各位小伙伴还可以获得uni-app入门视频和本专栏同步视频作为额外奖励。 小编目前在做毕业设计，主题为“高考志愿信息交流平台”，面向高中生和大学生，辛苦各位读者大佬朋友们填下问卷，点击链接https://www.wjx.cn/jq/98944127.aspx或扫描二维码、微信小程序码均可，希望各位能提供一些调查数据，先在这里谢过各位了(*^_^*)
前言
本文开始进入后端API开发部分，主要介绍了API环境搭建和登录API开发，具体包括以下几部分： 后端API环境搭建，包括后端线上环境部署、Postman的安装和使用、Python编辑器和数据库管理、项目创建和数据库配置等； API开发准备，包括数据表设计、异常类封装、用户验证实现和基础列表封装等； 登录API开发，包括手机验证码API开发、对接第三方短信发送平台、手机号登录API开发、账号密码登录API开发、第三方登录API开发、auth装饰器使用和退出登录API开发。
一、后端API环境搭建
1.后端线上环境部署
后端测试环境可以选择云服务器，也可以选择本地测试。 云服务器测试需要在云服务器提供商（阿里云、腾讯云等）购买云服务器，系统一般选择CentOS，配置可以根据自己的需要选择。 如果个人使用，也可以参与各大厂商推出的优惠活动，基本可以满足个人使用。
现以百度智能云https://cloud.baidu.com/为例，其2021新春活动https://cloud.baidu.com/campaign/2021-carnival/index.html的云服务器入门型 ic3为例演示云服务器的购买过程，如下：
如有需要购买用于个人使用的小伙伴可选择百度云或者华为云服务器，点击华为云服务器优惠链接或扫描下方二维码即可享受采购季优惠价：
购买成功之后，访问https://console.bce.baidu.com/bcc/#/bcc/instance/list，即可查看刚刚购买的实例，也可以修改管理员密码用于远程登录，如下：
此时点击VNC远程即可远程连接控制，一般用户名为root，密码为之前设置的管理员密码。
也可以使用XShell等工具连接控制。
还可以使用宝塔面板https://www.bt.cn/管理云服务器，实现更加方便的服务器管理，其免费版安装过程如下：
安装过程中，如果需要选择输入，直接输入y即可，安装完成后会给出管理面板访问地址和用户名密码，直接访问地址并使用提供的用户名密码访问即可。
如果不能访问，可能是因为云服务器未设置安全组、打开8888端口，具体创建和关联步骤可参考https://cloud.baidu.com/doc/BCC/s/6karx4ka3，配置常见的端口如下：
第一次登录到宝塔界面后，会提示推荐安装LNMP，根据如下配置（或自己选择）安装即可：
为了提高安全，可以在面板设置中设置面板端口、安全入口、授权IP等，如下：
为了在宝塔管理面板中创建站点，需要填写域名，可以在域名服务商处购买域名并进行实名认证和备案申请，然后进行域名解析、绑定服务器IP，以百度智能云为例，域名的选购、实名认证、解析和备案可查看文档https://cloud.baidu.com/doc/BCD/s/Wjwvymgtu。
之后即可在宝塔管理面板中创建站点，示例如下：
2.Postman安装使用
后端开发需要使用API调试管理工具，一般为Postman，可点击https://www.postman.com/downloads/下载。
下载完成之后安装即可。
Postman可以模拟各种方法的网络请求，包括GET、POST、PUT等，如下：
可以创建Colloection（相当于项目），用于保存一个项目的所有API请求，如下：
还可以创建环境变量，用于定义请求域名等，如下：
在整个项目完成后，还可以自动生成接口文档，免去了手动生成的麻烦。
3.PyCharm和数据库管理
后端API采用Django Restful Framework实现，主要的开发工具是PyCharm，可点击https://www.jetbrains.com/pycharm/download/#section=windows选择合适的系统和版本下载安装，同时需要安装Python3。
开发模式选择现在本地开发调试、开发完成之后再通过宝塔上传项目并配置线上运行的方式，因此下面所说的环境搭建和配置如无特别说明，都是在本地进行配置。
为了与其他Python运行环境区别开，建议创建虚拟环境来运行API服务，先通过pip install pipenv安装虚拟环境管理包，再在需要创建API项目的目录下执行pipenv shell来在当前目录下创建并进入虚拟环境，创建成功之后会输出creator CPython3Windows(dest=XXX\.virtualenvs\Uniapp_Practice_Community_Dating_App-7Br5cmK0, clear=False, global=False)，dest参数的值即为虚拟环境的位置，此时PyCharm中配置InterPreter为该目录下的Scripts目录下的python.exe即可选择编译器为所创建的虚拟环境。
为了项目所需，需要安装一些常见的第三方库，包括django、djangorestframework、markdown、django-filter等，直接在当前虚拟环境下执行pip install 库名 -i https://pypi.douban.com/simple即可。
如对Python虚拟环境的创建和使用有不清楚的地方，可参考https://blog.csdn.net/CUFEECR/article/details/108173573、https://blog.csdn.net/CUFEECR/article/details/108919863等文章。
数据库管理工具可以使用Navicat，可以方便地进行关系型数据库管理，包括数据库连接、数据操作、数据同步和备份等等，界面如下：
4.创建项目
在当前项目目录下执行django-admin startproject Community_Dating_API即可成功创建项目，进入Community_Dating_API目录后再执行python manage.py runserver即可运行项目，访问http://127.0.0.1:8000/，出现如下界面说明配置无误：
需要在项目主目录下New一个Python Package为apps，用于保存项目中所有的app； 并创建extra_apps包，用于保存源码经修改的第三方包。 同时为了以后开发更加方便，可以将apps和extra_apps设置为Sources Root。
Django的配置文件为项目目录下的Community_Dating_API目录下的settings.py，配置路径如下：
import os
import sys
# Build paths inside the project like this: os.path.join(BASE_DIR, ...)
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, BASE_DIR)
sys.path.insert(0, os.path.join(BASE_DIR, 'apps'))
sys.path.insert(0, os.path.join(BASE_DIR, 'extra_apps'))5.数据库创建和配置
本地MySQL安装可以使用集成工具PhpStudy，具体安装方法可参考https://blog.csdn.net/CUFEECR/article/details/107432591。
需要配置搜索引擎为InnoDB，如下：
然后开启MySQL服务。
可以命令行连接数据库，也可以使用Navicat，连接用户名和密码一般均为root，创建数据库名为Community_Dating，字符集为utf-8即可。
以命令行为例，创建数据库命令为create database Community_Dating;。
此时再使用Navicat执行.sql文件创建数据表，示例如下：
可以看到，成功创建了数据表。
如需
.sql文件和其他文件进行测试学习，可以直接点击加QQ群963624318 ，在群文件夹uni-app实战之社区交友APP中下载即可。
.sql文件分为带数据和不带数据两种，可根据需要使用。
然后需要进行Django数据库配置，需要安装数据库引擎mysqlclient，并配置settings.py如下：
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.mysql',
'NAME': 'community_dating',
'HOST': '127.0.0.1',
'PORT': 3306,
'USER': 'root',
'PASSWORD': 'root'
}
}再执行python manage.py inspectdb将MySQL数据库表生成对应的ORM模型，如下：
from django.db import models
class Adsense(models.Model):
src = models.CharField(max_length=255)
url = models.CharField(max_length=255, blank=True, null=True)
type = models.PositiveIntegerField()
create_time = models.DateTimeField(blank=True, null=True)
class Meta:
managed = False
db_table = 'adsense'
class Blacklist(models.Model):
black_id = models.PositiveIntegerField()
user_id = models.PositiveIntegerField()
create_time = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'blacklist'
class Comment(models.Model):
user_id = models.PositiveIntegerField()
fid = models.PositiveIntegerField()
fnum = models.PositiveIntegerField()
data = models.CharField(max_length=225)
create_time = models.IntegerField(blank=True, null=True)
post_id = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'comment'
class Feedback(models.Model):
to_id = models.PositiveIntegerField()
from_id = models.PositiveIntegerField()
data = models.CharField(max_length=255, blank=True, null=True)
create_time = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'feedback'
class Follow(models.Model):
follow_id = models.PositiveIntegerField()
user_id = models.PositiveIntegerField()
create_time = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'follow'
class Image(models.Model):
url = models.CharField(max_length=255)
create_time = models.IntegerField(blank=True, null=True)
user_id = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'image'
class Post(models.Model):
user_id = models.PositiveIntegerField()
title = models.CharField(max_length=80)
titlepic = models.CharField(max_length=255)
content = models.TextField()
sharenum = models.PositiveIntegerField()
path = models.CharField(max_length=255)
type = models.PositiveIntegerField()
create_time = models.IntegerField(blank=True, null=True)
post_class_id = models.IntegerField(blank=True, null=True)
share_id = models.PositiveIntegerField(blank=True, null=True)
isopen = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'post'
class PostClass(models.Model):
classname = models.CharField(max_length=5)
status = models.PositiveIntegerField()
create_time = models.DateTimeField(blank=True, null=True)
class Meta:
managed = False
db_table = 'post_class'
class PostImage(models.Model):
post_id = models.PositiveIntegerField()
image_id = models.PositiveIntegerField()
create_time = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'post_image'
class Support(models.Model):
user_id = models.PositiveIntegerField()
post_id = models.PositiveIntegerField()
type = models.PositiveIntegerField()
create_time = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'support'
class Topic(models.Model):
title = models.CharField(max_length=80)
titlepic = models.CharField(max_length=255)
desc = models.CharField(max_length=255)
type = models.PositiveIntegerField()
create_time = models.DateTimeField(blank=True, null=True)
topic_class_id = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'topic'
class TopicClass(models.Model):
classname = models.CharField(max_length=5)
status = models.PositiveIntegerField()
create_time = models.DateTimeField(blank=True, null=True)
class Meta:
managed = False
db_table = 'topic_class'
class TopicPost(models.Model):
topic_id = models.PositiveIntegerField()
post_id = models.PositiveIntegerField()
create_time = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'topic_post'
class Update(models.Model):
url = models.CharField(max_length=255, blank=True, null=True)
version = models.CharField(max_length=10, blank=True, null=True)
status = models.IntegerField(blank=True, null=True)
create_time = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'update'
class User(models.Model):
username = models.CharField(max_length=80)
userpic = models.CharField(max_length=255, blank=True, null=True)
password = models.CharField(max_length=255)
phone = models.CharField(max_length=11, blank=True, null=True)
email = models.CharField(max_length=255, blank=True, null=True)
status = models.PositiveIntegerField()
create_time = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'user'
class UserBind(models.Model):
type = models.CharField(max_length=50)
openid = models.CharField(max_length=255)
user_id = models.PositiveIntegerField(blank=True, null=True)
nickname = models.CharField(max_length=50, blank=True, null=True)
avatarurl = models.CharField(max_length=255, blank=True, null=True)
class Meta:
managed = False
db_table = 'user_bind'
class Userinfo(models.Model):
user_id = models.PositiveIntegerField()
age = models.PositiveIntegerField()
sex = models.PositiveIntegerField()
qg = models.PositiveIntegerField()
job = models.CharField(max_length=10, blank=True, null=True)
path = models.CharField(max_length=255, blank=True, null=True)
birthday = models.CharField(max_length=20, blank=True, null=True)
class Meta:
managed = False
db_table = 'userinfo'
在Community_ Dating_API目录下新建models.py用于临时保存这些ORM模型，再清空数据库、便于后面重新生成数据。
二、API开发准备
1.数据表设计
登录主要涉及到user、userinfo和user_bind表。 使用账号密码进行登录时，账号可以是昵称、手机号或邮箱，因此user表中也包含了这些字段，并通过正则表达式进行判断，user表字段和含义如下： 字段|含义 -----|----- id|主键 username|用户名 userpic|头像 password|密码 phone|手机号码 email|邮箱 create_ time|创建时间 status|状态，0禁用、1启用
userinfo表字段和含义如下： 字段|含义 -----|----- id|主键 user_id|用户Id sex|性别 age|年龄 qg|情感 job|工作 birthday|生日 path|家乡
手机验证码登录涉及到获取验证码和登录两个接口； 第三方登录也是使用接口实现。
user_bind表是第三方登录表，字段和含义如下： 字段|含义 -----|----- id|主键 type|第三方类型(微信、微博、QQ、其他) openid|第三方openid userid|用户id (默认是0) nickname|第二方呢称 avatarurl|第三方头像
当用户第一次使用第三方登录时，就会在user_bind表中添加记录，同时会绑定手机号，在user表中添加记录。
2.封装异常类
很多时候请求时会出现异常，但是显示到页面中的样式可能并不是我们所需要的，因为是接口开发，直接输出错误状态码和错误信息即可，这样可以统一接口的返回方式，即使视图函数执行出错也能被捕捉。
先执行命令python manage.py startapp user创建user app，来处理用户相关的逻辑，同时将生成的user目录移动至apps目录下；
将之前Community_Dating_API目录下models.puy中的User、UserBind和Userinfo3个模型移动至user/models.py中，如下：
from django.db import models
# Create your models here.
class User(models.Model):
username = models.CharField(max_length=80)
userpic = models.CharField(max_length=255, blank=True, null=True)
password = models.CharField(max_length=255)
phone = models.CharField(max_length=11, blank=True, null=True)
email = models.CharField(max_length=255, blank=True, null=True)
status = models.PositiveIntegerField(blank=True, null=True)
create_time = models.IntegerField(blank=True, null=True)
class Meta:
managed = False
db_table = 'user'
class UserBind(models.Model):
type = models.CharField(max_length=50)
openid = models.CharField(max_length=255)
user_id = models.PositiveIntegerField(blank=True, null=True)
nickname = models.CharField(max_length=50, blank=True, null=True)
avatarurl = models.CharField(max_length=255, blank=True, null=True)
class Meta:
managed = False
db_table = 'user_bind'
class Userinfo(models.Model):
user_id = models.PositiveIntegerField()
age = models.PositiveIntegerField(default='保密')
sex = models.PositiveIntegerField(default='保密')
qg = models.PositiveIntegerField(default='保密')
job = models.CharField(max_length=10, blank=True, null=True)
path = models.CharField(max_length=255, blank=True, null=True)
birthday = models.CharField(max_length=20, blank=True, null=True)
class Meta:
managed = False
db_table = 'userinfo'并依次执行python manage.py makemigrations和python manage.py migrate映射数据库。
user/views.py如下：
from rest_framework import mixins, viewsets
from .models import User
from .serializers import UserSerializer
# Create your views here.
class UserListViewSet(mixins.ListModelMixin, viewsets.GenericViewSet):
'''用户列表页'''
try:
queryset = User.objects.all()
serializer_class = UserSerializer
except:
pass
user目录下新建serializers.py如下：
from rest_framework import serializers
class UserSerializer(serializers.Serializer):
username = serializers.CharField(required=True, max_length=50)
phone = serializers.CharField(max_length=11)
create_time = serializers.IntegerField()
user目录下新建urls.py用于保存本app下的路由，如下：
from django.urls import path
from .views import UserListViewSet
urlpatterns = [
path('', UserListViewSet.as_view({'get': 'list'}))
]
Community_Dating_API/urls.py中导入user app中的路由，如下：
from django.conf.urls import url
from django.urls import path, include
from django.contrib import admin
urlpatterns = [
url(r'^admin/', admin.site.urls),
path('user/', include('apps.user.urls')),
]settings.py中安装user和rest_framework app如下：
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'rest_framework',
'apps.user'
]此时访问http://127.0.0.1:8000/user/，显示：
说明配置成功，显示的是restful_framework提供的接口页面，同时提供了API和JSON两种页面。
此时再在apps下新建一个Python Package为utils作为工具目录，用来保存工具函数。 utils.py下自定义异常处理器common_exception_handler.py如下：
import logging
from rest_framework import status
from rest_framework.response import Response
from rest_framework.serializers import Serializer
from rest_framework.views import exception_handler
from .errors import ErrorCode
c_fmt = "[%(levelname)s]%(asctime)s %(filename)s.%(funcName)s():line %(lineno)d :\n%(message)s"
date_format = "%Y-%m-%d %H:%M:%S %a"
logging.basicConfig(level=logging.INFO, format=c_fmt, datefmt=date_format)
logger = logging.getLogger("drf")
class RestResponse(Response):
def __init__(self, msg=None, success=None, data=None, status=None, errorCode=ErrorCode.SERVER_ERROR,
template_name=None, headers=None,
exception=False, content_type=None):
super(RestResponse, self).__init__(self, status=status)
if isinstance(data, Serializer):
msg = (
'You passed a Serializer instance as data, but '
'probably meant to pass serialized `.data` or '
'`.error`. representation.'
)
raise AssertionError(msg)
self.msg = msg
self.errorCode = errorCode
self.data = {"data": data or [], "msg": msg or "", "errorCode": errorCode}
self.template_name = template_name
self.exception = exception
self.content_type = content_type
if headers:
for name, value in headers.items():
self[name] = value
def common_exception_handler(exc, context):
response = exception_handler(exc, context)
context_view = context.get("view", None)
context_path = context.get('request').path
context_method = context.get('request').method
context_ip = context.get('request').META.get("REMOTE_ADDR")
if response is None:
data = {
'path': context_path,
'method': context_method,
'remote_address': context_ip
}
logger.error('%s,%s' % (context_view, exc))
response = RestResponse(msg=str(exc).replace('\\', ''), data=data, status=status.HTTP_500_INTERNAL_SERVER_ERROR, success=False)
return response
if response.status_code == 400:
response = RestResponse(data=response.data, status=response.status_code,
msg='400_bad request', success=False)
if response.status_code == 404:
response = RestResponse(data=response.data, status=response.status_code,
msg='404_Not Found', success=False)
if response.status_code == 401:
response = RestResponse(data=response.data, status=response.status_code,
msg='401_UNAUTHORIZED', success=False)
if response.status_code == 403:
response = RestResponse(data=response.data, status=response.status_code,
msg='403_FORBIDDEN ', success=False)
if response.status_code == 405:
response = RestResponse(data=response.data, status=response.status_code,
msg='405_METHOD_NOT_ALLOWED', success=False)
if 500 <= response.status_code <= 599:
response = RestResponse(data=response.data, status=response.status_code,
msg='INTERNAL_SERVER_ERROR', success=False)
return response可以看到，重写了DRF（Django Restful Framework的简写）提供的Response类。
同级目录下创建errors.py用于保存常见的异常类及相关常量，如下：
from rest_framework import status
from rest_framework.response import Response
class ErrorCode:
SERVER_ERROR = 999
UNAUTHORIZED = 10000
PAGE_NOT_FOUND = 10001
PERMISSION_DENIED = 10002
PARAM_ERROR = 40000
DATA_NOT_FOUND = 40001
DATA_NOT_VALID = 40002
REPEAT_POST = 40003
class ErrorMsg:
SERVER_ERROR = '服务器错误'
UNAUTHORIZED = '未登录'
PAGE_NOT_FOUND = '资源未找到'
PERMISSION_DENIED = '无权限'
PARAM_ERROR = '参数验证错误'
DATA_NOT_FOUND = '未找到数据'
DATA_NOT_VALID = '数据错误'
REPEAT_POST = '重复提交'
def ErrorResponse(code=status.HTTP_500_INTERNAL_SERVER_ERROR, msg=ErrorMsg.SERVER_ERROR, errorCode=ErrorCode.SERVER_ERROR, headers=None):
err = {
'code': code,
'msg': msg,
'errorCode': errorCode,
}
return Response(err, code, headers=headers)
class Error(Exception):
def __init__(self, code=status.HTTP_500_INTERNAL_SERVER_ERROR, msg=ErrorMsg.SERVER_ERROR, errorCode=ErrorCode.SERVER_ERROR):
self.code = code
self.msg = msg
self.errorCode = errorCode
def __str__(self):
return str(self.msg)
def getResponse(self):
return ErrorResponse(self.code, self.msg, self.errorCode)
class BaseException(Exception):
code = status.HTTP_500_INTERNAL_SERVER_ERROR
msg = ErrorMsg.SERVER_ERROR
errorCode = ErrorCode.SERVER_ERROR
def __init__(self, msg=None, errorCode=None):
if msg is None:
msg = self.msg
if errorCode is None:
errorCode = self.errorCode
def __str__(self):
return str(self.msg)
class ParamError(BaseException):
code = 400
class Unauthorized(BaseException):
code = 401
class PermissionDenied(BaseException):
code = 403
class ObjectNotFound(BaseException):
code = 404
class ServerError(BaseException):
code = 500
settings.py中添加配置如下：
# Restful Framework
REST_FRAMEWORK = {
'EXCEPTION_HANDLER': 'apps.utils.common_exception_handler.common_exception_handler'
}此时为了测试异常，在user/views.py中添加异常代码如下：
class UserListViewSet(mixins.ListModelMixin, viewsets.GenericViewSet):
'''用户列表页'''
try:
queryset = User.objects.all()
2/0
serializer_class = UserSerializer
except:
pass
再访问http://127.0.0.1:8000/user/，显示：
可以看到，显示了异常情况的返回数据。
3.用户验证实现
验证用户一般在新增或修改用户时会用到，新增一般使用POST方法，user/urls.py如下：
from django.urls import path
from .views import UserListViewSet
urlpatterns = [
path('', UserListViewSet.as_view({'get': 'list', 'post': 'create'}))
]
DRF中的serializers.py起到了Django中的forms.py的作用，用于验证模型字段，如下：
from rest_framework import serializers
from .models import User
class UserSerializer(serializers.ModelSerializer):
username = serializers.CharField(required=True, allow_blank=False, max_length=50, error_messages={'required': '用户名必填', 'blank': '用户名不能为空'})
email = serializers.EmailField(max_length=11, allow_blank=True, error_messages={'invalid': '邮箱格式不正确'})
create_time = serializers.IntegerField()
class Meta:
model = User
fields = '__all__'包含了所有字段。
views.py如下：
from rest_framework import mixins, viewsets
from rest_framework.mixins import ListModelMixin, CreateModelMixin
from .models import User
from .serializers import UserSerializer
# Create your views here.
class UserListViewSet(ListModelMixin, CreateModelMixin, viewsets.GenericViewSet):
'''用户列表页'''
try:
queryset = User.objects.all()
serializer_class = UserSerializer
except:
pass
因为UserListViewSet继承自CreateModelMixin，所以增加了添加用户的功能。
显示：
可以看到，显示了用户表的所有字段；
并且在使用POST方法添加时，如果字段验证失败，会给出相应的提示信息。
因为DRF提供了比较友好的前端交互页面，可以进行提交数据等操作，除此之外也可以使用Postman进行操作。
还可以不包括全部字段、而是指定一些字段，serializers.py如下：
from rest_framework import serializers
from .models import User
class UserSerializer(serializers.ModelSerializer):
username = serializers.CharField(required=True, allow_blank=False, max_length=50, error_messages={'required': '用户名必填', 'blank': '用户名不能为空'})
email = serializers.EmailField(max_length=11, allow_blank=True, error_messages={'invalid': '邮箱格式不正确'})
create_time = serializers.IntegerField()
class Meta:
model = User
fields = ['username', 'email', 'create_time']显示：
可以看到，此时只有3个字段。
4.封装基础列表
可以看到，之前返回的数据都是一个列表，列表中包含着数据对象，或者可以嵌套列表。 但是我们一般需要返回的JSON数据格式可能如下：
{
"code": 0,
"data": [],
"msg": "",
"total": ""
}此时可以在每个视图中继承ListModelMixin的同时重写list()方法。
同时，因为很多数据都存在这样的情况，因此我们可以选择实现一个函数来重定义数据格式，并在每次重写list()方法时调用该函数即可。
在utils目录下新建base_mixin.py用来保存经过重写的mixin方法或类，如下：
from rest_framework.response import Response
def base_list(view_set, code=0, msg='success'):
queryset = view_set.filter_queryset(view_set.get_queryset())
serializer = view_set.get_serializer(queryset, many=True)
response = {'code': code, 'data': serializer.data, 'msg': msg, 'total': len(serializer.data)}
return Response(response)
views.py如下：
from rest_framework import viewsets
from rest_framework.mixins import ListModelMixin, CreateModelMixin
from apps.utils.base_mixin import base_list
from .models import User
from .serializers import UserSerializer
# Create your views here.
class UserListViewSet(ListModelMixin, CreateModelMixin, viewsets.GenericViewSet):
'''用户列表页'''
try:
queryset = User.objects.all()
serializer_class = UserSerializer
except:
pass
def list(self, request, *args, **kwargs):
return base_list(self)
显示：
可以看到，实现了自定义的数据格式，以后其他数据列表也可以直接调用base_list()方法即可。
三、登录API开发和完善
1.手机验证码API开发
获取验证码流程如下: （1）接收手机号码； （2）验证手机号码合法性； （3）判断是否已经获取过验证码（判断缓存中是否存在当前手机号的验证码，有则进行提示）； （4）生成4位数随机数字； （5）发送短信（云片、容联云、阿里大于等平台）； （6）将手机号和验证码保存在缓存中（60秒）； （7）提示成功。
user/serializers.py中增加短信发送序列化如下：
class SmsSerializer(serializers.Serializer):
'''短信发送序列化'''
mobile = serializers.CharField(max_length=11, min_length=11, required=True,
error_messages={'max_length': '手机号长度过长', 'min_length': '手机号长度过短',
'required': '手机号必填', 'invalid': '手机号码不合法', 'blank': '手机号不能为空'})
def validate_mobile(self, mobile):
'''验证手机号码'''
# 验证手机号码是否合法
if not re.match(REGEX_MOBILE, mobile):
raise serializers.ValidationError('手机号格式有误，请重新输入')
return mobileutils目录下新建正则表达式常量库regex_rule.py如下：
# 手机号码验证正则表达式
REGEX_MOBILE = '^1[35789]\d{9}$|^147\d{8}$'utils目录下新建base_mixin.py用于保存重定义list和create返回样式，如下：
from rest_framework import status
from rest_framework.response import Response
def base_list(view_set, code=status.HTTP_200_OK, msg='success', error_code=None):
queryset = view_set.filter_queryset(view_set.get_queryset())
serializer = view_set.get_serializer(queryset, many=True)
response = {'code': code, 'data': serializer.data, 'msg': msg, 'total': len(serializer.data)}
if error_code:
response['errorCode'] = error_code
return Response(response, status=code)
def base_create(code=status.HTTP_200_OK, data=None, msg='success', error_code=None):
response = {'code': code, 'msg': msg}
if data:
response['data'] = data
response['total'] = len(data) if isinstance(data, (list, dict)) else 1
if error_code:
response['errorCode'] = error_code
return Response(response, status=code)
views.py中增加短信发送视图，如下：
class SmsCodeViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''发送短信验证码'''
serializer_class = SmsSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取手机号
mobile = serializer.validated_data['mobile']
redis_conn = get_redis_connection('verify_code')
sms_code_flag = redis_conn.get('sms_flag_'+mobile)
# 判断是否已经发送过
if sms_code_flag:
return base_create(msg=ErrorMsg.SEND_CODE_TOO_FAST, code=status.HTTP_400_BAD_REQUEST, error_code=ErrorCode.SEND_CODE_TOO_FAST)
# 生成4位验证码
code = generate_code()
sms_status = {'code': 0, 'msg': '发送成功', 'count': 1, 'fee': 0.05, 'unit': 'RMB', 'mobile': '13312345678', 'sid': 56592475448}
if sms_status['code'] != 0:
logger.error(sms_status)
return base_create(msg=ErrorMsg.SEND_FAILED, code=status.HTTP_400_BAD_REQUEST, error_code=ErrorCode.SEND_FAILED)
# 发送成功、写入缓存
else:
logger.info(sms_status)
redis_conn.setex('sms_'+mobile, SMS_CODE_EXPIRES, code)
redis_conn.setex('sms_flag_'+mobile, SMS_FLAG_EXPIRES, 1)
return base_create(msg=ErrorMsg.SEND_SUCCESS, error_code=ErrorCode.SEND_SUCCESS, data=code)
urls.py配置如下：
from django.urls import path
from .views import UserListViewSet, SmsCodeViewSet, PhoneLoginViewSet
urlpatterns = [
path('', UserListViewSet.as_view({'get': 'list', 'post': 'create'})),
path('sendcode/', SmsCodeViewSet.as_view({'post': 'create'})),
]
可以看到，通过Redis缓存保存短信验证码和发送标志。
utils目录下新建randowm_code.py保存生成验证码函数如下：
from random import choice
def generate_code(num=4):
'''生成指定位数验证码'''
seeds = '1234567890'
random_str = choice('123456789')
for i in range(num - 1):
random_str += choice(seeds)
return random_str
需要安装django_redis，直接执行pip install django_redis -i https://pypi.douban.com/simple命令即可。
除此之外，还需要在本地安装Redis并开启服务，可点击https://download.csdn.net/download/CUFEECR/12260885下载并解压，再配置环境变量、开启服务。
settings.py中配置Redis缓存，如下：
# Redis config
CACHES = {
"default": { # default
"BACKEND": "django_redis.cache.RedisCache",
"LOCATION": "redis://127.0.0.1:6379/0",
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
}
},
"verify_code": { # verify_code
"BACKEND": "django_redis.cache.RedisCache",
"LOCATION": "redis://127.0.0.1:6379/1",
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
}
},
}
自己约定，settings.py中保存与Django、DRF直接相关的配置项，其他配置（包括过期时间、APIKEY等常量）保存到另一个配置文件中，可以在utils目录下新建config.py，用于保存常量配置项，如下：
import logging
# logger config
c_fmt = "[%(levelname)s]%(asctime)s %(filename)s.%(funcName)s():line %(lineno)d :\n%(message)s"
date_format = "%Y-%m-%d %H:%M:%S %a"
logging.basicConfig(level=logging.INFO, format=c_fmt, datefmt=date_format)
logger = logging.getLogger("drf")
# SMS config
SMS_CODE_EXPIRES = 300
SMS_FLAG_EXPIRES = 60配置了日志和短信过期时间。
errors.py中补充常见的错误状态码和错误信息如下：
class ErrorCode:
SERVER_ERROR = 999
PARAM_ERROR = 10000
PAGE_NOT_FOUND = 10001
PERMISSION_DENIED = 10002
USER_NOT_FOUND = 20000
USER_FORBIDDEN = 20001
PASSWORD_ERROR = 20002
ILLEGAL_TOKEN = 20003
HAVE_QUITED = 20004
BIND_TYPE_CONFLICT = 20005
HAVE_BOUND = 20006
PHONE_HAVE_BOUND = 20007
NOT_BOUND = 20008
MODIFY_PASSWORD_FAILED = 20009
SEND_CODE_ERROR = 30000
SEND_CODE_TOO_FAST = 30001
INVALID_PHONE = 30002
REACH_DAY_LIMIT = 30003
SEND_FAILED = 30004
MSG_NOT_OPEN = 30005
REGET_MSG_CODE = 30006
CODE_ERROR = 30007
SEND_SUCCESS = 30008
FORBID_REPEAT = 40000
DATA_NOT_FOUND = 40001
DATA_NOT_VALID = 40002
REPEAT_POST = 40003
class ErrorMsg:
SERVER_ERROR = '服务器错误'
PARAM_ERROR = '通用参数错误'
PAGE_NOT_FOUND = '资源未找到'
PERMISSION_DENIED = '无权限'
USER_NOT_FOUND = '用户不存在'
USER_FORBIDDEN = '用户被禁用'
PASSWORD_ERROR = '密码错误'
ILLEGAL_TOKEN = '非法Token'
HAVE_QUITED = '立即退出'
BIND_TYPE_CONFLICT = '绑定类型冲突'
HAVE_BOUND = '已经被绑定'
PHONE_HAVE_BOUND = '就是当前手机号，无需再绑'
NOT_BOUND = '请先绑定手机'
MODIFY_PASSWORD_FAILED = '修改密码失败'
SEND_CODE_ERROR = '发送验证码错误'
SEND_CODE_TOO_FAST = '发送验证码过快，请稍后再试'
INVALID_PHONE = '无效号码'
REACH_DAY_LIMIT = '触发日限制'
SEND_FAILED = '发送失败' # 默认错误
MSG_NOT_OPEN = '未开启发生短信'
REGET_MSG_CODE = '请重新获取验证码'
CODE_ERROR = '验证码错误'
SEND_SUCCESS = '发送成功'
FORBID_REPEAT = '请勿重复操作' # 通用
DATA_NOT_FOUND = '未找到数据'
DATA_NOT_VALID = '数据错误'
REPEAT_POST = '重复提交'此时，进行测试如下：
可以看到，实现了对手机号码格式和发送频率的限制。
2.对接第三方短信发送平台
短信发送功能使用第三方短信服务提供商，包括云片、容联云、阿里大于等平台，这里以云片为例进行说明。 云片的账号注册、认证、签名和模板报备可参考https://blog.csdn.net/CUFEECR/article/details/106941804第四部分。
一般将发送短信封装为类，并调用类中提供的方法，如果在短时间内会有多个人同时发送，就需要多次实例化并调用方法，这会增加内存占用，因此可以使用单例化来减少内存占用。 在utils目录下创建yunpian.py如下：
import requests
from apps.utils.config import YUNPIAN_SEND_URL, YUNPIAN_API_KEY, SMS_CODE_EXPIRES
class YunPian(object):
def __new__(cls, *args, **kwargs):
if not hasattr(cls, '_instance'):
cls._instance = super().__new__(cls, *args, **kwargs)
cls._instance.single_send_url = YUNPIAN_SEND_URL
cls._instance.api_key = YUNPIAN_API_KEY
return cls._instance
def send_sms(self, code, mobile, timeout=SMS_CODE_EXPIRES // 60):
params = {
'apikey': self._instance.api_key,
'mobile': mobile,
'text': '【Python进化讲堂】您的验证码是{}。有效期为{}分钟，请尽快验证。'.format(code, timeout)
}
response = requests.post(self._instance.single_send_url, data=params).json()
return response
if __name__ == '__main__':
print(YunPian().send_sms('1234', '13312345678'))
config.py中配置如下：
# yunpian config
YUNPIAN_SEND_URL = 'https://sms.yunpian.com/v2/sms/single_send.json'
YUNPIAN_API_KEY = 'xxxx1361381f31b3957beda37fxxxxxx'views.py中完善短信发送如下：
class SmsCodeViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''发送短信验证码'''
serializer_class = SmsSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取手机号
mobile = serializer.validated_data['mobile']
redis_conn = get_redis_connection('verify_code')
sms_code_flag = redis_conn.get('sms_flag_'+mobile)
# 判断是否已经发送过
if sms_code_flag:
return base_create(msg=ErrorMsg.SEND_CODE_TOO_FAST, code=status.HTTP_400_BAD_REQUEST, error_code=ErrorCode.SEND_CODE_TOO_FAST)
# 生成4位验证码
code = generate_code()
sms_status = YunPian().send_sms(code, mobile)
# 发送成功、写入缓存
if sms_status['code'] == 0:
logger.info(sms_status)
redis_conn.setex('sms_' + mobile, SMS_CODE_EXPIRES, code)
redis_conn.setex('sms_flag_' + mobile, SMS_FLAG_EXPIRES, 1)
return base_create(msg=ErrorMsg.SEND_SUCCESS, error_code=ErrorCode.SEND_SUCCESS, data=code)
# 无效号码或其他参数
elif sms_status['code'] == 1:
return base_create(msg=ErrorMsg.INVALID_PHONE, error_code=ErrorCode.INVALID_PHONE, data=sms_status['msg'], code=status.HTTP_400_BAD_REQUEST)
# 触发日限制
elif sms_status['code'] == 43:
return base_create(msg=ErrorMsg.REACH_DAY_LIMIT, error_code=ErrorCode.REACH_DAY_LIMIT, data=sms_status['msg'],
code=status.HTTP_400_BAD_REQUEST)
# 其他发送失败情况
else:
logger.error(sms_status)
if settings.DEBUG:
return base_create(msg=sms_status['msg'], code=status.HTTP_400_BAD_REQUEST,
error_code=sms_status['code'])
else:
return base_create(msg=ErrorMsg.SEND_FAILED, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.SEND_FAILED)
显示：
可以看到，此时短信验证码发送成功。
再查看Redis，如下：
127.0.0.1:6379[1]> select 1
OK
127.0.0.1:6379[1]> keys *
1) "sms_flag_13188888888"
2) "sms_13188888888"
127.0.0.1:6379[1]> get sms_flag_13188888888
"1"
127.0.0.1:6379[1]> get sms_13188888888
"8522"
127.0.0.1:6379[1]>可以看到，与前台获取到的数据一致，验证码保存成功。
3.手机号登录API开发
手机号登录是在获取到短信验证码并验证成功之后，再判断是否是新用户，不存在则创建新用户，存在则判断是否是被禁用户。
最后将生成的Token保存到Redis缓存。
其中，生成Token使用的是djangorestframework-jwt，需要先执行pip install djangorestframework-jwt命令进行安装。
serializers.py定义验证如下：
class PhoneLoginSerializer(serializers.ModelSerializer):
'''用户序列化'''
code = serializers.CharField(max_length=4, min_length=4, required=True, label='验证码', help_text='输入4位验证码',
error_messages={
'required': '验证码验证',
'max_length': '请输入4位验证码',
'min_length': '请输入4位验证码',
'invalid': '验证码格式有误'
})
phone = serializers.CharField(max_length=11, min_length=11, required=True, label='手机号', help_text='输入11位手机号',
error_messages={'max_length': '手机号长度过长', 'min_length': '手机号长度过短',
'required': '手机号必填', 'invalid': '手机号码不合法', 'blank': '手机号不能为空'})
def validate_code(self, code):
if not re.match(REGEX_CODE_FOUR, code):
raise serializers.ValidationError('验证码格式有误，请重新输入')
return code
def validate(self, attrs):
# 验证验证码是否存在
code = attrs['code']
phone = attrs['phone']
redis_conn = get_redis_connection('verify_code')
sms_code_server = redis_conn.get('sms_' + phone)
if not sms_code_server:
raise serializers.ValidationError(ErrorMsg.REGET_MSG_CODE)
sms_code_server = sms_code_server.decode()
if sms_code_server != code:
print(sms_code_server, code)
print(type(sms_code_server), type(code))
raise serializers.ValidationError(ErrorMsg.CODE_ERROR)
del attrs['code']
attrs['username'] = attrs['phone']
return attrs
class Meta:
model = User
fields = ['phone', 'code']
在utils目录下新建data_handle.py，并定义根据查询参数查询用户的方法如下：
import re
from apps.user.models import User
from apps.utils.regex_rule import REGEX_MOBILE, EMAIL_RULE
def user_exists(param):
if isinstance(param, str):
queryset = User.objects.filter(username=param)
elif 'phone' in param:
queryset = User.objects.filter(phone=param['phone'])
elif 'email' in param:
queryset = User.objects.filter(email=param['email'])
else:
queryset = User.objects.filter(username=param)
if queryset:
return queryset.first()
else:
return None
views.py定义手机验证码业务逻辑如下：
from django.conf import settings
from django_redis import get_redis_connection
from rest_framework import viewsets, status
from rest_framework.mixins import ListModelMixin, CreateModelMixin
from rest_framework_jwt.serializers import jwt_encode_handler, jwt_payload_handler
from apps.utils.base_mixin import base_list, base_create
from apps.utils.config import SMS_CODE_EXPIRES, SMS_FLAG_EXPIRES, logger, TOKEN_EXPIRATION_DELTA
from apps.utils.data_handle import user_exists
from apps.utils.errors import ErrorCode, ErrorMsg
from apps.utils.random_code import generate_code
from .models import User, UserInfo
from .serializers import UserSerializer, SmsSerializer, PhoneLoginSerializer
redis_conn = get_redis_connection('verify_code')
# Create your views here.
class UserListViewSet(ListModelMixin, CreateModelMixin, viewsets.GenericViewSet):
'''用户列表页'''
try:
queryset = User.objects.all()
serializer_class = UserSerializer
except:
pass
def list(self, request, *args, **kwargs):
return base_list(self)
class SmsCodeViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''发送短信验证码'''
serializer_class = SmsSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取手机号
phone = serializer.validated_data['phone']
sms_code_flag = redis_conn.get('sms_flag_' + phone)
# 判断是否已经发送过
if sms_code_flag:
return base_create(msg=ErrorMsg.SEND_CODE_TOO_FAST, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.SEND_CODE_TOO_FAST)
# 生成4位验证码
code = generate_code()
# sms_status = YunPian().send_sms(code, phone)
sms_status = {'code': 0, 'msg': '发送成功', 'count': 1, 'fee': 0.05, 'unit': 'RMB', 'mobile': '13311111111',
'sid': 61811583014}
# 发送成功、写入缓存
if sms_status['code'] == 0:
logger.info(sms_status)
redis_conn.setex('sms_' + phone, SMS_CODE_EXPIRES, code)
redis_conn.setex('sms_flag_' + phone, SMS_FLAG_EXPIRES, 1)
return base_create(msg=ErrorMsg.SEND_SUCCESS, error_code=ErrorCode.SEND_SUCCESS, data=code)
# 无效号码或其他参数
elif sms_status['code'] == 1:
return base_create(msg=ErrorMsg.INVALID_PHONE, error_code=ErrorCode.INVALID_PHONE, data=sms_status['msg'],
code=status.HTTP_400_BAD_REQUEST)
# 触发日限制
elif sms_status['code'] == 43:
return base_create(msg=ErrorMsg.REACH_DAY_LIMIT, error_code=ErrorCode.REACH_DAY_LIMIT,
data=sms_status['msg'],
code=status.HTTP_400_BAD_REQUEST)
# 其他发送失败情况
else:
logger.error(sms_status)
if settings.DEBUG:
return base_create(msg=sms_status['msg'], code=status.HTTP_400_BAD_REQUEST,
error_code=sms_status['code'])
else:
return base_create(msg=ErrorMsg.SEND_FAILED, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.SEND_FAILED)
class PhoneLoginViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''手机号登录'''
serializer_class = PhoneLoginSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取手机号
phone = serializer.validated_data['phone']
# 验证用户是否存在
user = user_exists({'phone': phone})
if user:
user_status = user.status
# 判断用户是否被禁用
if user_status == 0:
return base_create(msg=ErrorMsg.USER_FORBIDDEN, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.USER_FORBIDDEN)
# 用户不存在、直接创建并注册
else:
# 保存用户
user = User.objects.create(username=phone, phone=phone)
# 在用户信息表创建记录
UserInfo.objects.create(user_id=user)
# 登录成功、返回Token
payload = jwt_payload_handler(user)
token = jwt_encode_handler(payload)
redis_conn.setex('token_' + phone, TOKEN_EXPIRATION_DELTA, token)
return base_create(msg='登录成功', code=status.HTTP_201_CREATED, data=token,
error_code=ErrorCode.USER_FORBIDDEN)
config.py配置如下：
# login config
TOKEN_EXPIRATION_DELTA = 7 * 24 * 60 * 60urls.py配置如下：
from django.urls import path
from .views import UserListViewSet, SmsCodeViewSet, PhoneLoginViewSet
urlpatterns = [
path('', UserListViewSet.as_view({'get': 'list', 'post': 'create'})),
path('sendcode/', SmsCodeViewSet.as_view({'post': 'create'})),
path('phonelogin/', PhoneLoginViewSet.as_view({'post': 'create'})),
]
修改user/models.py如下：
import time
from django.db import models
# Create your models here.
class User(models.Model):
username = models.CharField(max_length=80)
userpic = models.CharField(max_length=255, blank=True, null=True)
password = models.CharField(max_length=255)
phone = models.CharField(max_length=11)
email = models.CharField(max_length=255, blank=True, null=True)
status = models.PositiveIntegerField(default=1)
create_time = models.IntegerField(default=int(time.time()))
is_delete = models.BooleanField(default=False)
class Meta:
db_table = 'user'
class UserBind(models.Model):
type = models.CharField(max_length=50)
openid = models.CharField(max_length=255)
user_id = models.PositiveIntegerField(blank=True, null=True)
nickname = models.CharField(max_length=50, blank=True, null=True)
avatarurl = models.CharField(max_length=255, blank=True, null=True)
is_delete = models.BooleanField(default=False)
class Meta:
db_table = 'user_bind'
class UserInfo(models.Model):
user_id = models.OneToOneField(User, on_delete=models.CASCADE)
age = models.PositiveIntegerField(default=0)
sex = models.PositiveIntegerField(default=0)
qg = models.PositiveIntegerField(default=0)
job = models.CharField(max_length=10, blank=True, null=True)
path = models.CharField(max_length=255, blank=True, null=True)
birthday = models.CharField(max_length=20, blank=True, null=True)
is_delete = models.BooleanField(default=False)
class Meta:
db_table = 'userinfo'
并重新执行命令映射数据库。
regex_rule.py中定义短信验证码正则表达式如下：
# 短信验证码正则表达式
REGEX_CODE_FOUR = '^\d{4}$'显示：
可以看到，只有在手机号码和验证码验证成功之后，才会发送成功，并返回Token。
4.账号密码登录API开发
serializers.py中创建用户名登录序列化验证如下：
class LoginSerializer(serializers.ModelSerializer):
'''用户名密码登录序列化'''
username = serializers.CharField(max_length=15, min_length=4, required=True, label='用户名', help_text='输入4-15位用户名',
error_messages={
'required': '用户名必填',
'max_length': '用户名过长',
'min_length': '用户名过短'
})
password = serializers.CharField(max_length=18, min_length=6, required=True, label='密码', help_text='输入6-18位密码',
error_messages={'max_length': '密码过长', 'min_length': '密码过短',
'required': '密码必填', 'invalid': '密码不合法', 'blank': '密码不能为空'})
def validate_password(self, password):
if not re.match(PASSWORD_RULE, password):
raise serializers.ValidationError('密码格式有误，请重新输入')
return password
class Meta:
model = User
fields = ['username', 'password']对用户名和密码进行了验证。
views.py中定义登录视图如下：
class PhoneLoginViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''手机号登录'''
serializer_class = PhoneLoginSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取手机号
phone = serializer.validated_data['phone']
# 验证用户是否存在
user = user_exists({'phone': phone})
if user:
user_status = user.status
# 判断用户是否被禁用
if user_status == 0:
return base_create(msg=ErrorMsg.USER_FORBIDDEN, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.USER_FORBIDDEN)
# 用户不存在、直接创建并注册
else:
# 保存用户
user = User.objects.create(username=phone, phone=phone, password=make_password(phone[-6:]))
# 在用户信息表创建记录
UserInfo.objects.create(user_id=user)
# 登录成功、返回Token
payload = jwt_payload_handler(user)
token = jwt_encode_handler(payload)
redis_conn.setex('token_' + phone, TOKEN_EXPIRATION_DELTA, token)
return base_create(msg='登录成功', code=status.HTTP_201_CREATED, data=token,
error_code=ErrorCode.USER_FORBIDDEN)
class LoginViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''账号密码登录'''
serializer_class = LoginSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取账号和密码
username = serializer.validated_data['username']
password = serializer.validated_data['password']
# 验证用户是否存在
user = user_exists(filter_username(username))
if not user:
return base_create(msg=ErrorMsg.USER_NOT_FOUND, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.USER_NOT_FOUND)
# 判断用户是否被禁用
if user.status == 0:
return base_create(msg=ErrorMsg.USER_FORBIDDEN, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.USER_FORBIDDEN)
# 验证密码
if not self.check_password(password, user.password):
return base_create(msg=ErrorMsg.PASSWORD_ERROR, code=status.HTTP_201_CREATED,
error_code=ErrorCode.PASSWORD_ERROR)
# 登录成功、返回Token
payload = jwt_payload_handler(user)
token = jwt_encode_handler(payload)
redis_conn.setex('token_' + username, TOKEN_EXPIRATION_DELTA, token)
return base_create(msg='登录成功', code=status.HTTP_201_CREATED, data=token,
error_code=ErrorCode.USER_FORBIDDEN)
def check_password(self, raw_password, password):
if not password:
return False
if not check_password(raw_password, password):
return False
return True
可以看到，在之前定义的PhoneLoginViewSet中增加了验证码登录时保存密码字段，并设置默认密码为手机号后6位。
data_handle.py中完善用户查询如下：
import re
from apps.user.models import User
from apps.utils.regex_rule import REGEX_MOBILE, EMAIL_RULE
def user_exists(param):
if isinstance(param, str):
queryset = User.objects.filter(username=param)
elif 'phone' in param:
queryset = User.objects.filter(phone=param['phone'])
elif 'email' in param:
queryset = User.objects.filter(email=param['email'])
elif 'id' in param:
queryset = User.objects.filter(id=param['id'])
else:
queryset = User.objects.filter(username=param)
if queryset:
return queryset.first()
else:
return None
def filter_username(username):
param_dict = {}
# 验证是否是手机号码
if re.match(REGEX_MOBILE, username):
param_dict['phone'] = username
# 验证是否是邮箱
elif re.match(EMAIL_RULE, username):
param_dict['email'] = username
# 参数是用户名
else:
param_dict['username'] = username
return param_dict
可以看到，增加了判断传入的查询参数是用户名、手机号还是邮箱的函数。
urls.py中增加路由如下：
from django.urls import path
from .views import UserListViewSet, SmsCodeViewSet, PhoneLoginViewSet, LoginViewSet
urlpatterns = [
path('', UserListViewSet.as_view({'get': 'list', 'post': 'create'})),
path('sendcode/', SmsCodeViewSet.as_view({'post': 'create'})),
path('phonelogin/', PhoneLoginViewSet.as_view({'post': 'create'})),
path('login/', LoginViewSet.as_view({'post': 'create'})),
]
regex_rule.py中增加密码和邮箱正则表达式，如下：
# 密码正则表达式
PASSWORD_RULE = '^[a-zA-Z0-9_-]{6,18}$'
# 邮箱正则表达式
EMAIL_RULE = '^[a-zA-Z0-9_-]+@[a-zA-Z0-9_-]+(\.[a-zA-Z0-9_-]+)+$'显示：
可以看到，实现了用账号和密码进行登录。
5.第三方登录API开发
APP第三方登录可以使用uni-app提供的接口。 第三方登录包括微信、QQ和微博等，需要使用user_bind表，即UserBind，为了优化数据表结构，将其与User模型独立，具有多对一的关系，通过外键进行关联，如下：
class UserBind(models.Model):
type = models.CharField(max_length=10)
openid = models.CharField(max_length=100)
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='binds')
nickname = models.CharField(max_length=50, blank=True, null=True)
avatarurl = models.CharField(max_length=150, blank=True, null=True)
is_delete = models.BooleanField(default=False)
class Meta:
db_table = 'user_bind'并执行命令进行数据库映射。
实现时，先在user_bind表中创建记录，再通过user_id绑定到User表中相应的用户。 如果用户第一次登录选择的是第三方登录时，默认绑定的user_id为-1，需提前在user表中手动添加一条id为-1的记录，因为Django不支持添加id为负值的记录； 用户要进行交互式操作时，强制其绑定手机号码，即在user表创建用户，并将其id作为当前user_bind表的user_id，从而可以实现一个用户对应一个user表记录，同时可以对应多个第三方登录账号。
先定义序列化验证如下：
class OtherLoginSerializer(serializers.ModelSerializer):
'''第三方登录序列化'''
type = serializers.CharField(required=True, error_messages={'required': '类型必填'})
openid = serializers.CharField(required=True, error_messages={'required': 'openid必填'})
nickname = serializers.CharField(required=True, error_messages={'required': '昵称必填'})
avatarurl = serializers.CharField(required=True, error_messages={'required': '用户头像必填'})
expires_in = serializers.IntegerField(required=True, error_messages={'required': '有效期必填'})
class Meta:
model = UserBind
fields = ['type', 'openid', 'nickname', 'avatarurl', 'expires_in']
def validate(self, attrs):
del attrs['expires_in']
return attrs
views.py中定义视图如下：
from django.conf import settings
from django.contrib.auth.hashers import make_password, check_password
from django_redis import get_redis_connection
from rest_framework import viewsets, status
from rest_framework.mixins import ListModelMixin, CreateModelMixin
from apps.utils.base_mixin import base_list, base_create
from apps.utils.config import SMS_CODE_EXPIRES, SMS_FLAG_EXPIRES, logger, PK_MAXIMUM
from apps.utils.data_handle import user_exists, filter_username, create_save_token, check_status
from apps.utils.errors import ErrorCode, ErrorMsg
from apps.utils.random_code import generate_code
from .models import User, UserInfo, UserBind
from .serializers import UserSerializer, SmsSerializer, PhoneLoginSerializer, LoginSerializer, OtherLoginSerializer
redis_conn = get_redis_connection('verify_code')
# Create your views here.
class UserListViewSet(ListModelMixin, CreateModelMixin, viewsets.GenericViewSet):
'''用户列表页'''
try:
queryset = User.objects.all()
serializer_class = UserSerializer
except:
pass
def list(self, request, *args, **kwargs):
return base_list(self)
class SmsCodeViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''发送短信验证码'''
serializer_class = SmsSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取手机号
phone = serializer.validated_data['phone']
sms_code_flag = redis_conn.get('sms_flag_' + phone)
# 判断是否已经发送过
if sms_code_flag:
return base_create(msg=ErrorMsg.SEND_CODE_TOO_FAST, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.SEND_CODE_TOO_FAST)
# 生成4位验证码
code = generate_code()
# sms_status = YunPian().send_sms(code, phone)
sms_status = {'code': 0, 'msg': '发送成功', 'count': 1, 'fee': 0.05, 'unit': 'RMB', 'mobile': '13311111111',
'sid': 61811583014}
# 发送成功、写入缓存
if sms_status['code'] == 0:
logger.info(sms_status)
redis_conn.setex('sms_' + phone, SMS_CODE_EXPIRES, code)
redis_conn.setex('sms_flag_' + phone, SMS_FLAG_EXPIRES, 1)
return base_create(msg=ErrorMsg.SEND_SUCCESS, error_code=ErrorCode.SEND_SUCCESS, data=code)
# 无效号码或其他参数
elif sms_status['code'] == 1:
return base_create(msg=ErrorMsg.INVALID_PHONE, error_code=ErrorCode.INVALID_PHONE, data=sms_status['msg'],
code=status.HTTP_400_BAD_REQUEST)
# 触发日限制
elif sms_status['code'] == 43:
return base_create(msg=ErrorMsg.REACH_DAY_LIMIT, error_code=ErrorCode.REACH_DAY_LIMIT,
data=sms_status['msg'],
code=status.HTTP_400_BAD_REQUEST)
# 其他发送失败情况
else:
logger.error(sms_status)
if settings.DEBUG:
return base_create(msg=sms_status['msg'], code=status.HTTP_400_BAD_REQUEST,
error_code=sms_status['code'])
else:
return base_create(msg=ErrorMsg.SEND_FAILED, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.SEND_FAILED)
class PhoneLoginViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''手机号登录'''
serializer_class = PhoneLoginSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取手机号
phone = serializer.validated_data['phone']
# 验证用户是否存在
user = user_exists({'phone': phone})
# 用户不存在、直接创建并注册
if not user:
# 保存用户
user = User.objects.create(username=phone, phone=phone, password=make_password(phone[-6:]))
# 在用户信息表创建记录
UserInfo.objects.create(user_id=user.id)
# 判断用户是否被禁用
user_info = check_status(user)
# 登录成功、返回Token
token = create_save_token(user_info)
return base_create(msg='登录成功', code=status.HTTP_201_CREATED, data=token,
error_code=ErrorCode.USER_FORBIDDEN)
class LoginViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''账号密码登录'''
serializer_class = LoginSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取账号和密码
username = serializer.validated_data['username']
password = serializer.validated_data['password']
# 验证用户是否存在
user = user_exists(filter_username(username))
if not user:
return base_create(msg=ErrorMsg.USER_NOT_FOUND, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.USER_NOT_FOUND)
# 判断用户是否被禁用
user_info = check_status(user)
# 验证密码
if not self.check_password(password, user.password):
return base_create(msg=ErrorMsg.PASSWORD_ERROR, code=status.HTTP_201_CREATED,
error_code=ErrorCode.PASSWORD_ERROR)
# 登录成功、返回Token
token = create_save_token(user_info)
return base_create(msg='登录成功', code=status.HTTP_201_CREATED, data=token,
error_code=ErrorCode.USER_FORBIDDEN)
def check_password(self, raw_password, password):
if not password:
return False
if not check_password(raw_password, password):
return False
return True
class OtherLoginViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''第三方登录'''
serializer_class = OtherLoginSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取参数
validated_data = serializer.validated_data
provider = validated_data['type']
openid = validated_data['openid']
nickname = validated_data['nickname']
avatar_url = validated_data['avatarurl']
# 验证用户是否存在
user = user_exists({'provider': provider, 'openid': openid})
user_bind = None
if not user:
user = User.objects.filter(id=-1).first()
user_bind = UserBind.objects.create(type=provider, openid=openid, nickname=nickname, avatarurl=avatar_url,
user=user)
# 判断用户是否被禁用
user_info = check_status(user, True)
# 登录成功、返回Token
token = create_save_token(model_to_dict(user_bind) if user_bind else user_info)
return base_create(msg='登录成功', code=status.HTTP_201_CREATED, data=token,
error_code=ErrorCode.USER_FORBIDDEN)
data_handle.py中完善如下：
import re
from django.forms import model_to_dict
from django_redis import get_redis_connection
from apps.user.models import User, UserBind
from apps.utils.base_mixin import base_create
from apps.utils.config import TOKEN_EXPIRATION_DELTA
from apps.utils.errors import ErrorMsg, ErrorCode
from apps.utils.regex_rule import REGEX_MOBILE, EMAIL_RULE
from apps.utils.token_generater import create_token
redis_conn = get_redis_connection('verify_code')
def user_exists(param):
if isinstance(param, str):
queryset = User.objects.filter(username=param)
elif 'phone' in param:
queryset = User.objects.filter(phone=param['phone'])
elif 'email' in param:
queryset = User.objects.filter(email=param['email'])
elif 'id' in param:
queryset = User.objects.filter(id=param['id'])
elif 'provider' in param:
user_bind = UserBind.objects.filter(type=param['provider'], openid=param['openid'])
if user_bind:
queryset = user_bind.first().user
return queryset
else:
queryset = None
else:
queryset = User.objects.filter(username=param)
if queryset:
return queryset.first()
else:
return None
def filter_username(username):
param_dict = {}
# 验证是否是手机号码
if re.match(REGEX_MOBILE, username):
param_dict['phone'] = username
# 验证是否是邮箱
elif re.match(EMAIL_RULE, username):
param_dict['email'] = username
# 参数是用户名
else:
param_dict['username'] = username
return param_dict
def create_save_token(info):
expire = info['expires_in'] if info.get('expires_in') else TOKEN_EXPIRATION_DELTA
token = create_token(info)
info['token'] = token
redis_conn.setex(token, expire, str(info))
return token
def check_status(user, refresh_data=False):
info = model_to_dict(user)
if refresh_data:
user_id = info['user_id'] if info.get('user_id') else info['id']
if user_id < 1:
return info
user = User.objects.filter(id=user_id).first()
status = user.status
else:
status = info['status']
if status == 0:
return base_create(msg=ErrorMsg.USER_FORBIDDEN, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.USER_FORBIDDEN)
return info
可以看到，将判断用户是否存在、是否被禁用和创建token封装到函数中，并调用，降低了耦合性。
生成token不再使用djangorestframework-jwt，改用django.core.signing，utils.py下新建token_generator.py如下：
import hashlib
import time
from django.core import signing
from django_redis import get_redis_connection
HEADER = {'typ': 'JWP', 'alg': 'default'}
SALT = 'www.corleytown.cn'
redis_conn = get_redis_connection('verify_code')
def encrypt(obj):
"""加密"""
value = signing.dumps(obj, salt=SALT)
value = signing.b64_encode(value.encode()).decode()
return value
def decrypt(src):
"""解密"""
src = signing.b64_decode(src.encode()).decode()
raw = signing.loads(src, salt=SALT)
print(type(raw))
return raw
def create_token(userinfo):
"""生成token信息"""
# 1. 加密头信息
header = encrypt(HEADER)
# 2. 构造Payload
if isinstance(userinfo, str):
payload = {"username": userinfo, "iat": time.time()}
else:
payload = userinfo
payload.update({"iat": time.time()})
payload = encrypt(payload)
# 3. 生成签名
md5 = hashlib.md5()
md5.update(("%s.%s" % (header, payload)).encode())
signature = md5.hexdigest()
token = "%s.%s.%s" % (header, payload, signature)
return token
urls.py中增加路由定义如下：
from django.urls import path
from .views import UserListViewSet, SmsCodeViewSet, PhoneLoginViewSet, LoginViewSet, OtherLoginViewSet
urlpatterns = [
path('', UserListViewSet.as_view({'get': 'list', 'post': 'create'})),
path('sendcode/', SmsCodeViewSet.as_view({'post': 'create'})),
path('phonelogin/', PhoneLoginViewSet.as_view({'post': 'create'})),
path('login/', LoginViewSet.as_view({'post': 'create'})),
path('otherlogin/', OtherLoginViewSet.as_view({'post': 'create'})),
]
显示：
可以看到，也实现了对第三方登录账号进行加密。
6.auth装饰器使用
很多接口都需要验证用户是否有访问该接口的权限，然后再验证用户提交的参数。 例如，要访问退出登录接口，则首先需要判断用户是否已经登录，登录之后才能进行后续的操作。 要实现权限验证，可以使用中间件，也可以使用装饰器，这里选择装饰器。
在utils目录下新建api_decorator.py如下：
from rest_framework import status
from apps.utils.base_mixin import base_create
from apps.utils.config import REDIS_CONN
from apps.utils.errors import ErrorCode, ErrorMsg
def auth_decorator(func):
need_auth = ['/user/logout/', ]
def inner(view, request, *args, **kwargs):
request_path = request.path
if request_path in need_auth:
# 获取头部信息
headers = request.headers
# 不含token
if 'Token' not in headers:
return base_create(msg=ErrorMsg.ILLEGAL_TOKEN, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.ILLEGAL_TOKEN)
token = headers.get('Token')
# 判断当前用户是否存在
user = REDIS_CONN.get(token)
# 未登录或已过期
if not user:
return base_create(msg=ErrorMsg.ILLEGAL_TOKEN, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.ILLEGAL_TOKEN)
# 将token和user_id携带到request对象中
request.data['user_token'] = token
user = dict(user.decode())
request.data['user_id'] = user['user_id'] if user.get('type') else user['id']
request.data['user_token_userinfo'] = user
return func(view, request, *args, **kwargs)
return inner
config.py中新建配置如下：
# redis chche
REDIS_CONN = get_redis_connection('verify_code')创建登出页面（为了演示效果，与第三方登录页面相同，后面再完善）如下：
class LogoutViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''退出登录'''
serializer_class = OtherLoginSerializer
@auth_decorator
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取参数
validated_data = serializer.validated_data
provider = validated_data['type']
openid = validated_data['openid']
nickname = validated_data['nickname']
avatar_url = validated_data['avatarurl']
# 验证用户是否存在
user = user_exists({'provider': provider, 'openid': openid})
user_bind = None
if not user:
user = User.objects.get_or_create(id=PK_MAXIMUM)[0]
user_bind = UserBind.objects.create(type=provider, openid=openid, nickname=nickname, avatarurl=avatar_url,
user=user)
# 判断用户是否被禁用
user_info = check_status(user, True)
# 登录成功、返回Token
token = create_save_token(model_to_dict(user_bind) if user_bind else user_info)
return base_create(msg='登录成功', code=status.HTTP_201_CREATED, data=token,
error_code=ErrorCode.USER_FORBIDDEN)
urls.py如下：
from django.urls import path
from .views import UserListViewSet, SmsCodeViewSet, PhoneLoginViewSet, LoginViewSet, OtherLoginViewSet, LogoutViewSet
urlpatterns = [
path('', UserListViewSet.as_view({'get': 'list', 'post': 'create'})),
path('sendcode/', SmsCodeViewSet.as_view({'post': 'create'})),
path('phonelogin/', PhoneLoginViewSet.as_view({'post': 'create'})),
path('login/', LoginViewSet.as_view({'post': 'create'})),
path('otherlogin/', OtherLoginViewSet.as_view({'post': 'create'})),
path('logout/', LogoutViewSet.as_view({'post': 'create'})),
]
显示：
可以看到，header未携带token，因此不能访问，返回错误信息。
再实现验证当前用户是否绑定手机号，api_decorators.py如下：
def bind_phone_decorator(func):
need_bind_phone = ['/user/logout/', ]
def inner(view, request, *args, **kwargs):
request_path = request.path
if request_path in need_bind_phone:
user_token_userinfo = request.user_token_userinfo
print(user_token_userinfo)
user_id = other_login_bind_phone(user_token_userinfo)
if not isinstance(user_id, int):
return user_id
return func(view, request, *args, **kwargs)
return innerdata_handle.py中实现验证第三方登录是否绑定手机如下：
def other_login_bind_phone(user):
# 第三方登录
if 'type' in user:
if user['user'] >= PK_MAXIMUM:
return base_create(msg=ErrorMsg.NOT_BOUND, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.NOT_BOUND)
return user['user']
# 账号密码登录
return user['id']views.py如下：
class LogoutViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''退出登录'''
serializer_class = OtherLoginSerializer
@auth_decorator
@bind_phone_decorator
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取参数
validated_data = serializer.validated_data
provider = validated_data['type']
openid = validated_data['openid']
nickname = validated_data['nickname']
avatar_url = validated_data['avatarurl']
# 验证用户是否存在
user = user_exists({'provider': provider, 'openid': openid})
user_bind = None
if not user:
user = User.objects.get_or_create(id=PK_MAXIMUM)[0]
user_bind = UserBind.objects.create(type=provider, openid=openid, nickname=nickname, avatarurl=avatar_url,
user=user)
# 判断用户是否被禁用
user_info = check_status(user, True)
# 登录成功、返回Token
token = create_save_token(model_to_dict(user_bind) if user_bind else user_info)
return base_create(msg='登录成功', code=status.HTTP_201_CREATED, data=token,
error_code=ErrorCode.USER_FORBIDDEN)
显示：
可以看到，在请求头携带token访问退出登录接口时，提示绑定手机。
再实现验证当前用户是否被禁用，如下：
def user_status_decorator(func):
need_status = ['/user/logout/', ]
def inner(view, request, *args, **kwargs):
request_path = request.path
if request_path in need_status:
user_token_userinfo = request.user_token_userinfo
user = check_status(user_token_userinfo)
if not isinstance(user, dict):
return user
return func(view, request, *args, **kwargs)
return inner7.退出登录API开发
现完善退出登录API，如下：
from django.conf import settings
from django.contrib.auth.hashers import make_password, check_password
from django.forms import model_to_dict
from rest_framework import viewsets, status
from rest_framework.mixins import ListModelMixin, CreateModelMixin
from .models import User, UserInfo, UserBind
from .serializers import UserSerializer, SmsSerializer, PhoneLoginSerializer, LoginSerializer, OtherLoginSerializer
from ..utils.api_decorator import auth_decorator
from ..utils.base_mixin import base_list, base_create
from ..utils.config import SMS_CODE_EXPIRES, SMS_FLAG_EXPIRES, logger, REDIS_CONN
from ..utils.data_handle import user_exists, filter_username, create_save_token, check_status
from ..utils.errors import ErrorCode, ErrorMsg
from ..utils.random_code import generate_code
# Create your views here.
class UserListViewSet(ListModelMixin, CreateModelMixin, viewsets.GenericViewSet):
'''用户列表页'''
try:
queryset = User.objects.all()
serializer_class = UserSerializer
except:
pass
def list(self, request, *args, **kwargs):
return base_list(self)
class SmsCodeViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''发送短信验证码'''
serializer_class = SmsSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取手机号
phone = serializer.validated_data['phone']
sms_code_flag = REDIS_CONN.get('sms_flag_' + phone)
# 判断是否已经发送过
if sms_code_flag:
return base_create(msg=ErrorMsg.SEND_CODE_TOO_FAST, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.SEND_CODE_TOO_FAST)
# 生成4位验证码
code = generate_code()
# sms_status = YunPian().send_sms(code, phone)
sms_status = {'code': 0, 'msg': '发送成功', 'count': 1, 'fee': 0.05, 'unit': 'RMB', 'mobile': '13311111111',
'sid': 61811583014}
# 发送成功、写入缓存
if sms_status['code'] == 0:
logger.info(sms_status)
REDIS_CONN.setex('sms_' + phone, SMS_CODE_EXPIRES, code)
REDIS_CONN.setex('sms_flag_' + phone, SMS_FLAG_EXPIRES, 1)
return base_create(msg=ErrorMsg.SEND_SUCCESS, error_code=ErrorCode.SEND_SUCCESS, data=code)
# 无效号码或其他参数
elif sms_status['code'] == 1:
return base_create(msg=ErrorMsg.INVALID_PHONE, error_code=ErrorCode.INVALID_PHONE, data=sms_status['msg'],
code=status.HTTP_400_BAD_REQUEST)
# 触发日限制
elif sms_status['code'] == 43:
return base_create(msg=ErrorMsg.REACH_DAY_LIMIT, error_code=ErrorCode.REACH_DAY_LIMIT,
data=sms_status['msg'],
code=status.HTTP_400_BAD_REQUEST)
# 其他发送失败情况
else:
logger.error(sms_status)
if settings.DEBUG:
return base_create(msg=sms_status['msg'], code=status.HTTP_400_BAD_REQUEST,
error_code=sms_status['code'])
else:
return base_create(msg=ErrorMsg.SEND_FAILED, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.SEND_FAILED)
class PhoneLoginViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''手机号登录'''
serializer_class = PhoneLoginSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取手机号
phone = serializer.validated_data['phone']
# 验证用户是否存在
user = user_exists({'phone': phone})
# 用户不存在、直接创建并注册
if not user:
# 保存用户
user = User.objects.create(username=phone, phone=phone, password=make_password(phone[-6:]))
# 在用户信息表创建记录
UserInfo.objects.create(user_id=user.id)
# 判断用户是否被禁用
user_info = check_status(user)
if not isinstance(user_info, dict):
return user_info
# 登录成功、返回Token
token = create_save_token(user_info)
return base_create(msg='登录成功', code=status.HTTP_201_CREATED, data=token,
error_code=ErrorCode.USER_FORBIDDEN)
class LoginViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''账号密码登录'''
serializer_class = LoginSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取账号和密码
username = serializer.validated_data['username']
password = serializer.validated_data['password']
# 验证用户是否存在
user = user_exists(filter_username(username))
if not user:
return base_create(msg=ErrorMsg.USER_NOT_FOUND, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.USER_NOT_FOUND)
# 判断用户是否被禁用
user_info = check_status(user)
if not isinstance(user_info, dict):
return user_info
# 验证密码
if not self.check_password(password, user.password):
return base_create(msg=ErrorMsg.PASSWORD_ERROR, code=status.HTTP_201_CREATED,
error_code=ErrorCode.PASSWORD_ERROR)
# 登录成功、返回Token
token = create_save_token(user_info)
return base_create(msg='登录成功', code=status.HTTP_201_CREATED, data=token,
error_code=ErrorCode.USER_FORBIDDEN)
def check_password(self, raw_password, password):
if not password:
return False
if not check_password(raw_password, password):
return False
return True
class OtherLoginViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''第三方登录'''
serializer_class = OtherLoginSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 获取参数
validated_data = serializer.validated_data
provider = validated_data['type']
openid = validated_data['openid']
nickname = validated_data['nickname']
avatar_url = validated_data['avatarurl']
# 验证用户是否存在
user = user_exists({'provider': provider, 'openid': openid})
user_bind = None
if not user:
user = User.objects.filter(id=-1).first()
user_bind = UserBind.objects.create(type=provider, openid=openid, nickname=nickname, avatarurl=avatar_url,
user=user)
# 判断用户是否被禁用
user_info = check_status(user, True)
if not isinstance(user_info, dict):
return user_info
# 登录成功、返回Token
token = create_save_token(model_to_dict(user_bind) if user_bind else user_info)
return base_create(msg='登录成功', code=status.HTTP_201_CREATED, data=token,
error_code=ErrorCode.USER_FORBIDDEN)
class LogoutViewSet(CreateModelMixin, viewsets.GenericViewSet):
'''退出登录'''
@auth_decorator
def create(self, request, *args, **kwargs):
if REDIS_CONN.delete(request.user_token):
return base_create(msg=ErrorMsg.LOGOUT_SUCCESS, code=status.HTTP_200_OK,
error_code=ErrorCode.LOGOUT_SUCCESS)
else:
return base_create(msg=ErrorMsg.LOGOUT_FAILED, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.LOGOUT_FAILED)
data_handle.py如下：
import re
from django.forms import model_to_dict
from rest_framework import status
from apps.user.models import User, UserBind
from apps.utils.base_mixin import base_create
from apps.utils.config import TOKEN_EXPIRATION_DELTA, REDIS_CONN, PK_MAXIMUM
from apps.utils.errors import ErrorMsg, ErrorCode
from apps.utils.regex_rule import REGEX_MOBILE, EMAIL_RULE
from apps.utils.token_generater import create_token
# 判断用户是否存在
def user_exists(param):
if isinstance(param, str):
queryset = User.objects.filter(username=param)
elif 'phone' in param:
queryset = User.objects.filter(phone=param['phone'])
elif 'email' in param:
queryset = User.objects.filter(email=param['email'])
elif 'id' in param:
queryset = User.objects.filter(id=param['id'])
elif 'provider' in param:
user_bind = UserBind.objects.filter(type=param['provider'], openid=param['openid'])
if user_bind:
queryset = user_bind.first().user
return queryset
else:
queryset = None
else:
queryset = User.objects.filter(username=param)
if queryset:
return queryset.first()
else:
return None
# 判断用户名类型
def filter_username(username):
param_dict = {}
# 验证是否是手机号码
if re.match(REGEX_MOBILE, username):
param_dict['phone'] = username
# 验证是否是邮箱
elif re.match(EMAIL_RULE, username):
param_dict['email'] = username
# 参数是用户名
else:
param_dict['username'] = username
return param_dict
# 生成保存token
def create_save_token(info):
expire = info['expires_in'] if info.get('expires_in') else TOKEN_EXPIRATION_DELTA
token = create_token(info)
info['token'] = token
REDIS_CONN.setex(token, expire, str(info))
return token
# 检查用户状态
def check_status(user, refresh_data=False):
info = user
if not isinstance(user, dict):
info = model_to_dict(user)
if refresh_data:
user_id = info['user'] if info.get('user') else info['id']
if user_id < 1:
return info
user = User.objects.filter(id=user_id).first()
user_status = user.status
else:
user_status = info['status']
if user_status == 0:
return base_create(msg=ErrorMsg.USER_FORBIDDEN, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.USER_FORBIDDEN)
return info
# 验证第三方登录是否绑定手机
def other_login_bind_phone(user):
# 第三方登录
if 'type' in user:
if user['user'] < 1:
return base_create(msg=ErrorMsg.NOT_BOUND, code=status.HTTP_400_BAD_REQUEST,
error_code=ErrorCode.NOT_BOUND)
return user['user']
# 账号密码登录
return user['id']
显示：
可以看到，实现了退出登录的效果。
总结
后端API开发可以根据需要选择合适的语言和开发框架，这里选择的是Python语言和Django Restful Framework框架，这是一个组件丰富、并且可扩展的接口框架，可以在利用其原有特性的基础上进行合理扩展，以达到自己的业务需求。
本文原文首发来自博客专栏uni-app社区交友APP开发实战，由本人转发至https://www.helloworld.net/p/wG8I57fXKCgb，其他平台均属侵权，可点击https://blog.csdn.net/CUFEECR/article/details/113825216查看原文，也可点击https://blog.csdn.net/CUFEECR浏览更多优质原创内容。
