使用 Celery 异步操作
Celery是由Python开发、简单、灵活、可靠的分布式任务队列,是一个处理异步任务的框架,其本质是生产者消费者模型,生产者发送任务到消息队列,消费者负责处理任务。Celery侧重于实时操作,但对调度支持也很好,其每天可以处理数以百万计的任务。
由于 django 框架请求/响应的过程是同步的,框架本身无法实现异步响应。所以异步执行前端一般会使用 Ajax,后端则使用 Celery。另外 django-celery 插件已经有一段时间没有更新了,对于新版本的 python 和 django 会有适配问题,所以使用 celery
一些参考文档
celery 使用步骤
django-celery 使用步骤:
- 安装、配置
- 设置 celery 主体,创建应用对象
- 根据需要创建任务
- 数据库迁移,生成 celery 需要的数据表
- 调用任务,将其添加至消息队列
- 启动 celery,监听队列并执行任务
- 查看执行结果,对结果进行处理
安装及环境配置
首先安装 django-celery,因为 celery 本身不实现中间件 Broker 的功能,所以还需要使用中间件。django 和 redis 配合的不错,这里就使用 django-redis
pip install django-celery django-redis
这里需要注意的是,windows 在 celery4.0 之后不支持多进程方式,而是更换成了协程方式,所以要使用 eventlet 或 gevent。
pip install eventlet
然后在 settings.py 中添加配置信息
##### celery 配置 #####
from urllib.parse import quote
PASSWORD = quote('123456') #使用有特殊字符密码,带有特殊字符需要进行转换才能识别
# Broker配置,使用Redis作为消息中间件
BROKER_URL = f'redis://mast:{
PASSWORD}@127.0.0.1:6379/8'.format(PASSWORD) # 此格式为连接需要验证的 redis
# 后台结果,如果没有此参数则使用 orm 的数据库
CELERY_RESULT_BACKEND = f'redis://mast:{
PASSWORD}@127.0.0.1:6379/9'.format(PASSWORD)
# 结果序列化方案
CELERY_RESULT_SERIALIZER = 'json'
# 任务结果过期时间,秒
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24
# 指定导入的任务模块,可以指定多个
CELERY_IMPORTS = ('app.tasks',) # 参数为 tasks.py 文件路径(即 应用.tasks)
CELERY_TIMEZONE = ’Asia/Shanghai' # 设置时区
CELERYD_LOG_FILE = BASE_DIR + "/logs/celery/celery.log" # log路径
CELERYBEAT_LOG_FILE = BASE_DIR + "/logs/celery/beat.log" # beat log路径
# 设置定时器策略
from datetime import timedelta
CELERYBEAT_SCHEDULE = {
# 定时任务一
u'邮件发送': {
# 任务名称
'task': 'app.tasks.print_now', # 需要执行的任务函数
# 'schedule': crontab(minute='*/2'), # 延迟
'schedule': timedelta(seconds=5), # 间隔5秒
'args': ('现在时间',), # 参数
},
}
# schedule 参数是执行频率,可以是整型(秒)、timedelta对象、crontab对象
# 还可以设置 kwargs 字典型的关键字参数
# 对于 crontab 函数使用举例
# crontab(hour='*/24') 每隔24小时
# crontab(minute=30, hour=0) 每天的凌晨 00:30
# crontab(hour=6, minute=0, day_of_month='1') 每月1号的 6:00
celery 主体
在主工程目录(settings.py 所在目录)添加 celery.py,主体代码就写在这里
import os
from celery import Celery
from django.conf import settings
# 设置项目运行的环境变量 DJANGO_SETTINGS_MODULE
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'DAdmin.settings') # 将 DAdmin.settings 添加到环境变量,根据工程名称改变
# 创建 celery 应用对象
app = Celery('AdminCelery')
# 加载配置
app.config_from_object('django.conf:settings')
# 如果在工程的应用中创建了tasks.py模块,那么celery应用会自动去添加任务
# 比如添加了一个任务,在 django 中会实时地检索出来
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
然后需要在项目中加载 celery 的 app,在项目主工程目录的 init.py 中,添加 all 属性
from .celery import app as celery_app #引入 celery.py 里的 app 对象
# 向项目模块中增加 celery_app 对象
__all__ = ['celry_app']
创建任务 tasks
每个任务本质上就是一个函数,写在 tasks.py 中。
import datetime
from celery import shared_task
import logging
logger = logging.getLogger(__name__)
# 定时任务,在 settings.py 的 CELERYBEAT_SCHEDULE 中注册
# 参数 info 是在注册时传入
# 调用时也可以触发
@shared_task
def print_now(info):
print(info, datetime.datetime.now())
# 也可以输出日志
logger.info(info, datetime.datetime.now())
# 没有在 CELERYBEAT_SCHEDULE 中注册,只能通过调用触发
@shared_task
def add(x, y):
time.sleep(5)
return x + y
迁移数据库
celery 的迁移和 django 其他的一样
python manage.py makemigrations
python manage.py migrate
启动 celery
接下来就是启动 celery 服务 worker,用来获取消息和执行任务
celery -A django_test worker -l info
mac、linux 系统使用此命令,-A 指定工程项目,-l 指定日志等级为 info
celery -A django_test worker -l info -P eventlet
windows 使用此命令,-A 指定工程项目,-P 指定执行单元使用 eventlet 实现后台异步操作,-l 指定日志等级为 info
然后是 beat,启动后就能够执行定时任务了
celery -A django_test beat -l info
在视图函数中调用任务和获取结果
非定时任务必须进行调用
from .tasks import add # 引入写好的任务
def add(request): # 视图函数
...
task = add.delay(100, 200) # 执行任务,返回任务异步结果对象
return HttpResponse(json.dumps({
'status': 'ok','task_id': task.task_id}),'application/json')
当调用时立刻能获得异步结果对象,此时对象的状态为 PENDING ,结果为 None。
该对象的一些常用的属性和方法:
- state 任务状态
- task_id 任务id
- result 任务结果
- ready() 判断任务是否结束
- wait(t) 等待t秒后获取结果,若任务执行完毕,则不等待直接获取结果,若任务在执行中,则wait期间一直阻塞,直到超时报错
- successful() 判断任务是否成功,成功为True,否则为False
通常获取该对象的 task_id 然后可以在另一函数中使用 task_id 获取任务状态和结果
from celery import result
def get_result_by_taskid(request):
task_id = request.GET.get('task_id')
# 获取异步结果对象
ar = result.AsyncResult(task_id)
if ar.ready():
return JsonResponse({
'status': ar.state, 'result': ar.result})
else:
return JsonResponse({
'status': ar.state, 'result': ''})
关于结果的存储
celery 的结果可以使用 django-celery-results 包来方便的保存至数据库
任务绑定
Celery可通过task绑定到实例获取到task的上下文,这样我们可以在task运行时候获取到task的状态,记录相关日志等
from celery import shared_task
import logging
logger = logging.getLogger(__name__)
# 任务绑定
@shared_task(bind=True) # bind=True 设置任务绑定
def add(self,x, y): # 第一个参数 self 能获取任务实例对象
try:
logger.info('add__-----'*10)
logger.info('name:',self.name)
logger.info('dir(self)',dir(self))
raise Exception
except Exception as e:
# 出错每4秒尝试一次,总共尝试4次
self.retry(exc=e, countdown=4, max_retries=4)
return x + y
任务钩子
Celery在执行任务时,提供了钩子方法用于在任务执行完成时候进行对应的操作,在Task源码中提供了很多状态钩子函数如:on_success(成功后执行)、on_failure(失败时候执行)、on_retry(任务重试时候执行)、after_return(任务返回时候执行)
from celery import Task
class MyHookTask(Task):
def on_success(self, retval, task_id, args, kwargs):
logger.info(f'task id:{
task_id} , arg:{
args} , successful !')
def on_failure(self, exc, task_id, args, kwargs, einfo):
logger.info(f'task id:{
task_id} , arg:{
args} , failed ! erros: {
exc}')
def on_retry(self, exc, task_id, args, kwargs, einfo):
logger.info(f'task id:{
task_id} , arg:{
args} , retry ! erros: {
exc}')
# 在对应的task函数的装饰器中,通过 base=MyHookTask 指定
@shared_task(base=MyHookTask, bind=True)
def add(self,x, y):
logger.info('add__-----'*10)
logger.info('name:',self.name)
logger.info('dir(self)',dir(self))
return x + y
flower
flower 是 celery 的一个图形化管理界面
富文本方案
django-mdeditor
DjangoUeditor
django-RESTful
REST 即表述性状态传递(Representational State Transfer),它是一种针对网络应用的设计和开发方式,可以降低开发的复杂性,提高系统的可伸缩性。REST通常基于使用HTTP,URI,和XML以及HTML这些现有的广泛流行的协议和标准。
RESTful API四大基本原则:
- 为每个资源设置URI
- 通过XML / JSON进行数据传递
- 无状态连接,服务器端不应保存过多上下文状态,即每个请求都是独立的
- 使用HTTP动词:GET POST PUT DELETE
安装和环境配置
django-RESTful 需要以下包支持(除了主插件程序包外,其他的包为可选项)
- DjangoRESTframework - 主插件程序包
- PyYAML, uritemplate (5.1+, 3.0.0+) - Schema生成支持。
- Markdown (3.0.0+) - 为browsable API 提供Markdown支持。
- Pygments (2.4.0+) - 为Markdown处理提供语法高亮。
- django-filter (1.0.1+) - Filtering支持。
- django-guardian (1.1.1+) - 对象级别的权限支持。
可以根据需要安装相应的包
pip install djangorestframework
pip install markdown # 为browsable API 提供Markdown支持。
pip install django-filter # Filtering支持。
使用 django-RESTful 需要注册应用到 settings.py 的 INSTALLED_APPS 中
INSTALLED_APPS = [
...
'rest_framework',
]
REST framework API的所有全局设定都会放在一个叫REST_FRAMEWORK的配置词典里,并添加到 settings.py 中:
REST_FRAMEWORK = {
# 在这里配置访问许可
'DEFAULT_PERMISSION_CLASSES': [
# 'rest_framework.permissions.DjangoModelPermissionsOrAnonReadOnly' # 匿名只读,登录用户可用
'rest_framework.permissions.IsAdminUser', # 只能管理员使用
],
# 配置分页器
'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
'PAGE_SIZE': 10,
}
如果打算用browsable API(一个可视化API测试工具),可能也会用REST framework的登录注销视图。可以添加路由到根目录的urls.py文件
urlpatterns = [
...
path('api-auth/', include('rest_framework.urls')) # 路由路径可以更改
]
测试安装和配置
以一个实例进行测试:创建一个读写API来访问项目的用户信息。
在根目录的 urls.py 中创建API
# urls.py
from django.urls import path, include
from django.contrib.auth.models import User
from rest_framework import routers, serializers, viewsets
# 序列化器是用来定义API的表示形式。
class UserSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = User
fields = ['url', 'username', 'email', 'is_staff']
# ViewSets定义视图的行为。
class UserViewSet(viewsets.ModelViewSet):
queryset = User.objects.all()
serializer_class = UserSerializer
# 路由器提供一个简单自动的方法来决定URL的配置。
router = routers.DefaultRouter()
router.register(r'users', UserViewSet) # 注册路由
# 通过URL自动路由来给我们的API布局。
# 此外,我们还要把登录的URL包含进来。
urlpatterns = [
path('', include(router.urls)), # 注册 REST 路由到主路由
path('api-auth/', include('rest_framework.urls', namespace='rest_framework'))
]
现在可以在浏览器中打开(默认页,即第一条路由)http://127.0.0.1:8000/,查看 ‘user’ API了。如果使用了右上角的登录控制,还可以在系统中添加、创建并删除用户。这样就是安装和配置成功了。
基本用法
在实际项目中,django-RESTful 的使用分为三个步骤:
- 创建序列化器:及定义API的表现形式
- 创建视图:用来处理api请求并返回响应
- 注册路由:将视图注册到路由中
创建序列化器
定义序列化程序,可以创建一个新的文件,这里使用 serializers.py
from django.contrib.auth.models import User
from rest_framework import serializers
# 定义序列化处理器,继承自 serializers.HyperlinkedModelSerializer,使用超链接关系
class UserSerializer(serializers.ModelSerializer):
class Meta:
# model 定义使用的数模型
model = User
# fields 定义使用数据模型的哪些字段
fields = ('url', 'username', 'email', 'groups')
django-RESTful 提供了3个序列化类的模板: Serializer、ModelSerializer、HyperlinkedModelSerializer
- Serializer : 标准序列化类。需要完全自定义声明相关字段,且需实现
create(**validate_data)
和update(instance, **validate_data)
两个函数 - ModelSerializer : Serialize 的子类,使用数据模型的序列化类。通过声明 Meta 类来指定相关属性(使用的模型类和需要序列化的字段)
- HyperlinkedModelSerializer : ModelSerializer 的子类,
创建视图
django-RESTful 提供了一些预设的视图处理类
from rest_framework import viewsets
from django.contrib.auth.models import User
from .serializers import UserSerializer
# 视图处理类
class UserViewSet(viewsets.ModelViewSet):
# queryset 为查询的结果集合
queryset = User.objects.all()
# serializer_classs 使用的序列化器
serializer_class = UserSerializer
创建路由
可以在应用中创建路由,并注册到主路由中
# 应用中的 urls.py
from django.urls import path, include
from rest_framework import routers
# 路由器提供一个简单自动的方法来决定URL的配置。
# 创建路由器
router = routers.DefaultRouter()
# 在路由器中注册路由及处理器(FBV或CBV)
router.register(r'users', UserViewSet)
# 通过URL自动路由来给我们的API布局。
urlpatterns = [
path('', include(router.urls)), # 将 REST 路由器中的路由添加到 app 的路由中
path('api-auth/', include('rest_framework.urls')), # 添加 browsable API 的登录路由
]
自定义 API 相关
根据实际情况自定义序列化器或者使用自定义的视图函数,有一些需要使用到的
序列化与反序列化
在自定义的视图中,需要将数据通过序列化器序列化后发出响应,接收到的请求也需要通过反序列化获取具体数据
from .serializers import UserSerializer
from django.contrib.auth.models import User
s1 = UserSerializer(User.objects.get(pk=1)) # 序列化单条数据
s2 = UserSerializer(User.objects.all(), many=True) # 序列化多条数据
# 渲染成Json字符串
from rest_framework.renderers import JSONRenderer
content = JSONRenderer().render(s1.data)
# 对已经渲染的Json字符串反序列化为Json对象
data = JSONParser().parse(io.BytesIO(content))
# 或直接解析 request 对象(POST)
# from rest_framework.parsers import JSONParser
# data = JSONParser().parse(request)
# serializer = UserSerializer(data=data) # 根据提交的参数获取序列化对象
# if serizlizer.is_valid(): # 通过验证
# serializer.save() # 保存数据
关联关系的序列化
在查看和测试API的过程中,会发现某些数据表的外键指向的是关联表的数据链接,而不是数据内容。如果要使用数据内容,就需要将关联关系序列化。在定义序列化器时,将外键字段声明成为字符串关系字段即可。
from django.contrib.auth.models import User
from rest_framework import serializers
# 定义序列化处理器,继承自 serializers.HyperlinkedModelSerializer,使用超链接关系
class UserSerializer(serializers.HyperlinkedModelSerializer):
# 在此定义字段
# 定义关系字段,将数据信息字符串化
groups = serializers.StringRelatedField() # 如果是对多关系(对方表是多端)则需要参数 many=True
# 也可以将关系对象整体序列化(即返回的 json 中,此字段是给包含了所有数据的 json 对象)
# groups = GroupSerializer() # 需先定义 GroupSerializer 序列化器,且如果对方是多端也需要 many=True
class Meta:
# model 定义使用的数模型
model = User
# fields 定义使用数据模型的哪些字段
fields = ('url', 'username', 'email', 'groups')
这个方法在一对一、一对多、多对多关系都适用。如果对方是多端(即此表的该外键字段或反查字段有多个数据),添加参数 many=True 即可。其中一些关联字段类型为
- StringRelatedField : 获取关联模型对象字符串化(使用 str() 函数返回的字符串)
- PrimarykeyRelatedField : 获取关联模型对象的主键
- SlugRelatedField : 获取关联模型对象指定字段(slug_field 参数值)
- HyperlinkedRelatedField : 获取关联模型对象的查询api链接,此种方式为默认方式
自定义视图处理器
使用的封装好的django-RESTful视图处理类能够方便的获取请求返回响应,但是如果需要使用自定义的视图处理器,则可使用 @api_view
装饰器(FBV)或 APIView
基类(CBV)。
FBV 方式举例,需注意的是注册路由使用 path 方法直接到 urlpatterns
from rest_framework.decorators import api_view
from .serializers import UserSerializer
from rest_framework.response import Response
from rest_framework.parsers import JSONParser
from django.contrib.auth.models import User
from django.http import JsonResponse
@api_view(['GET', 'POST'])
def user_func(request, pk=None):
if request.method == 'GET':
if not pk:
queryset = User.objects.all()
serializer = UserSerializer(queryset, many=True, context={
'request': request})
else:
queryset = User.objects.get(pk=pk)
serializer = UserSerializer(queryset, context={
'request': request})
# django-RESTful 重新封装了 Response,可以直接序列化
# 可以根据请求类型返回数据,例如浏览器直接请求会返回管理页面,请求 json 格式则返回 json 数据
# 也可以在请求地址后添加参数 ?format=json 指定获取 json 数据或 api 页面
return Response(serializer.data)
# return JsonResponse(serializer.data) # 返回的是序列化后的json字符串
elif request.method == 'POST':
# 接收的 request 是 django-RESTful 重新封装后的 request 对象
# 可以直接将其内容反序列化,然后通过序列化器从数据库中查询相关数据,再进行序列化
data = JSONParser().parse(request)
serializer = UserSerializer(data, context={
'request': request})
# 序列化器对象可以进行验证
if serializer.is_valid():
serializer.save() # 保存至数据库
return Response(serializer.data, status=201)
else:
return Response(serializer.errors, status=400)
CBV 方式举例,需注意的是注册路由使用 path 方法直接到 urlpatterns
from rest_framework.views import APIView
from .serializers import UserSerializer
from django.contrib.auth.models import User
from rest_framework.response import Response
from rest_framework.parsers import JSONParser
class UserClass(APIView):
def get(self, request, pk=None):
if not pk:
queryset = User.objects.all()
serializer = UserSerializer(queryset, many=True, context={
'request': request})
else:
queryset = User.objects.get(pk=pk)
serializer = UserSerializer(queryset, context={
'request': request})
return Response(serializer.data)
def post(self,request):
data = JSONParser().parse(request)
serializer = UserSerializer(data, context={
'request': request})
# 序列化器对象可以进行验证
if serializer.is_valid():
serializer.save() # 保存至数据库
return Response(serializer.data, status=201)
else:
return Response(serializer.errors, status=400)
获取 request 的数据
rest_framework 的 request 和 django 的 request 不太一样,获取数据是使用 request.data
,类似于 django 的 request.body
但是不是字节码,而是字符串。
name = request.data.get('name', None)
关于授权认证
默认情况下, APIView 中的相关接口方法不验证权限(授权),对资源并不安全,所以需要增加验证。
首先在 settings.py 中的 REST_FRAMEWORK 字段配置权限访问许可
# settings.py
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [ # 默认使用的授权认证
'rest_framework.authentication.BasicAuthentication', # 基本授权认证
'rest_framework.authentication.SessionAuthentication', # 基于session的授权认证
]
}
然后可以在视图中指定验证方式和许可类型
class EcampleView(APIView):
authentication_classes = (SessionAuthentication, BasicAuthentication) # 验证方式
permission_classes = (IsAuthenticated,) # 许可类型
def get(self, request):
pass
但是这种授权是基于 session 登录的,即 auth 模块的认证。在 api 接口中,通常会使用 token 进行认证。
django-RESTful 使用的 token 验证
TokenAuthentication 提供了简单的基于 Token 的HTTP认证方案,适用于客户端 - 服务器设置,如本地桌面和移动客户端。要在 django-RESTful 中使用 TokenAuthentication,需要配置认证类包含 TokenAuthentication,另外需要注册 rest_framework.authtoken 这个 app。需注意的是,确保在修改设置后运行一下 manage.py migrate
,因为 rest_framework.authtoken 会提交一些数据库迁移操作。
# settings.py
INSTALLED_APPS = [
...
'rest_framework.authtoken',
]
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [ # 默认使用的授权认证
'rest_framework.authentication.TokenAuthentication', # token授权认证
]
}
创建令牌
from rest_framework.authtoken.models import Token
from django.contrib.auth.models import User
user = User.objects.get(pk=1) # 获取用户
token = Token.objects.create(user=user) # 根据用户创建 token 实例
print(token.key) # token.key 就是需要验证的字段
通常会在每个用户创建时,创建对应的 token,可以捕捉用户的 post_save 信号
from django.conf import settings
from django.db.models.signals import post_save
from django.dispatch import receiver
from rest_framework.authtoken.models import Token
@receiver(post_save, sender=settings.AUTH_USER_MODEL)
def create_auth_token(sender, instance=None, created=False, **kwargs):
if created:
Token.objects.create(user=instance)
也可以为现有用户生成令牌
from django.contrib.auth.models import User
from rest_framework.authtoken.models import Token
for user in User.objects.all():
Token.objects.get_or_create(user=user)
# 此方法可以返回2个值,分别为 token 对象和 created
验证令牌
对客户端进行身份验证,token需要包含在名为 Authorization 的HTTP头中。密钥应该是以字符串"Token"为前缀,以空格分割的两个字符串。例如:
function ajax_get() {
fetch('/api/user/, {
headers: {
'Authorization': 'Token 9944b09199c62bcf9418ad846dd0e4bbdfc6ee4b'
}
}).then(response=>response.json())
.then(data=>{
console.log(data)
})
}
如果认证成功,TokenAuthentication 提供以下认证信息:
- request.user 将是一个Django User 实例。
- request.auth 将是一个rest_framework.authtoken.models.Token 实例。
那些被拒绝的未经身份验证的请求会返回使用适当WWW-Authenticate标头的HTTP 401 Unauthorized响应。例如:
WWW-Authenticate: Token
通过暴露 api 端点获取令牌
当使用TokenAuthentication时,可能希望为客户端提供一个获取给定用户名和密码的令牌的机制。 REST framework 提供了一个内置的视图来提供这个功能。要使用它,需要将 obtain_auth_token 视图添加到你的URLconf:
from rest_framework.authtoken import views
urlpatterns += [
path('api-token-auth/', views.obtain_auth_token) # url 路由可以自定义
]
当使用form表单或JSON将有效的username和password字段POST提交到视图时,obtain_auth_token 视图将返回JSON响应:
{
'token' : '9944b09199c62bcf9418ad846dd0e4bbdfc6ee4b' }
请注意,默认的obtain_auth_token视图显式使用JSON请求和响应,而不是使用settings中配置的默认渲染器和解析器类。如果需要自定义版本的obtain_auth_token视图,可以通过重写ObtainAuthToken类,并在url conf中使用它来实现。默认情况下,没有权限或限制应用于obtain_auth_token视图。如果你希望应用限制,则需要重写视图类,并使用throttle_classes属性包含它们。
邮件
django 有组件支持发送邮件
配置邮件信息
发送邮件需要配置邮件服务器信息
EMAIL_HOST = 'smtp.163.com'
EMAIL_PORT = 25
EMAIL_HOST_USER = '[email protected]'
EMAIL_HOST_PASSWORD = 'emailPassword'
需注意的是 EMAIL_HOST_PASSWORD 使用的是邮件服务器的授权码而不是登录密码。
发送邮件的方法
django 使用 django.core.mail.send_mail()
发送邮件,其用法为
from django.core.mail import send_mail
send_mail(title, message, from_email, recipient_list, fail_silently, auth_user, auth_password, connection, html_message)
- title : 邮件标题,字符串
- message : 邮件正文,字符串,必须有此参数,哪怕是空字符串。
- from_mail : 发送者,字符串
- recipient_list : 接收者,字符串列表,可以有多个接收者
- fail_silently :
- auth_user :
- auth_password :
- connection :
- html_message : 正文中的 html 文本,可以代替 message 参数,接受 html 内容
单元测试
可以使用 unittest 库进行单元测试。创建测试类,继承自 unittest 库中的相应类,然后创建方法。方法可以独立执行,用作测试使用。单元测试的方便之处在于可以直接调用已经写好的代码,或测试写好的代码。注意单元测试的方法名以test为开头,如果不以test开头则不会进行测试。
from django.test import TestCase
import unittest
class UserTestCase(TestCase):
def setUp(self):
print('--执行测试前进行配置--')
def test_01_add(self):
print('--add order--')
def test_02_get_info(self):
print('--info--')
def tearDown(self):
print('--测试后进行释放资源等收尾工作--')
if __name__ == '__main__':
unittest.main() # 执行单个测试
执行顺序为:
- setUp 用于初始化资源
- 各自定义方法,不是按照声明顺序,而是按照 ASCII 排序,所以常以 test 加序号加业务名称来声明
- tearDown 用于回收资源
单个套件
除了直接执行测试类,也可以创建套件执行
from unittest import TestSuite, TextTestRunner
def suite(): # 声明套件类
suite_ = TestSuite() # 创建测试套件对象
suite_.addTest(UserTestCase.test_01_add) # 添加套件的测试方法
suite_.addTest(UserTestCase.test_02_get_info)
return suite_
if __name__ == '__main__':
runner = TextTestRunner()
runner.run(suite())
多个套件
多个套件可以按照排序来进行不同的测试类的测试
def suite1(): # 套件类1
suite_ = TestSuite() # 创建测试套件对象
suite_.addTest(UserTestCase.test_01_add) # 添加套件的测试方法
suite_.addTest(UserTestCase.test_02_get_info)
return suite_
def suite2(): # 套件类2
suite_ = TestSuite() # 创建测试套件对象
suite_.addTest(OrderTestCase.test_01_add) # 添加套件的测试方法
suite_.addTest(OrderTestCase.test_02_get_info)
if __name__ == '__main__':
TextTestRunner().run(TestSuite((suite1(), suite2()))) # 按顺序进行测试
TestCase类的方法
- assertTrue(boolean condition)
如果 condition 为 false 则失败;否则通过测试; - assertEquals(Object expected, Object actual)
根据 equals() 方法,如果 expected 和 actual 不相等则失败;否则通过测试; - assertEquals(int expected, int actual)
根据==操作符,如果 expected 和 actual 不相等则失败;否则通过测试。对每一个原始类型:int、float、double、char、byte、long、short和boolean,这个方法都会都一个函数的重载。(参见assertEquals() 的注释) - assertSame(Object expected, Object actual)
如果 expected 和 actual 引用不同的内存对象则失败;如果它们引用相同的内存对象则通过测试。两个对象可能并不是相同的,但是它们可能通过 equals() 方法仍然可以是相等的 - assertNull(Object object)
如果对象为null则通过测试,反之看作失败
测试模型
django 测试模型不会使用实际数据库, 会为测试创建单独的空白数据库。所以进行模型测试需要先在 setUp()
方法中创建模型数据,进行数据初始化。当测试正常结束后,无论结果如何,会将临时创建的数据库销毁。
测试接口(视图、单元)
django 在TestCase对象中定义了Client类用于模拟客户端发送请求,所以可以使用这个工具来测试接口或视图。
def test_api(self):
response = self.client.post('/api')
self.assertEqual(r.status_code, 200)
content = json.loads(r.content)
self.assertEqual(content['result'], True)
client 可以使用 get 、post、put 等常用发放模拟客户端发送请求,这些方法的格式基本一致:post(path, data, content_type, follow, secure)
。其参数含义为:
- path 发送请求使用url
- data 发送请求时携带的数据
- content_type 携带数据的格式
- secure 客户端将模拟HTTPS请求
- follow 客户端将遵循任何重定向
测试 UI (模板)
线程共用和独立的静态数据
django 是基于多线程的,每一个请求处理都是一个独立的线程。当定义一个全局的静态数据时(python 没有真正意义上的静态数据,所定义的静态数据也是可以更改的,这里的静态只是相对而言),所有视图均可以访问此数据。因此静态数据是线程共用的,即不管访问用户和访问请求,均使用同一个数据。
如果希望有一个数据是线程独立的,例如需要在信号、钩子函数中访问 request 信息,而又不方便显式将 request 作为参数传递的时候。此数据在请求中间件随请求创建,随响应消亡。可以使用 threading
模块的 local()
方法来创建线程数据。
线程数据
使用 local()
方法将返回一个本地线程对象,此对象是线程独立的。可以将数据保存在此对象的属性中,在需要时调取。
# 自定义的中间件文件 middleware.py
from django.utils.deprecation import MiddlewareMixin
from threading import local
_locals = local()
# getter函数
def get_current_request():
return getattr(_locals,'request',None)
# 处理请求时添加数据
class RequestMiddlewart(MiddlewareMixin):
@staticmethod
def process_request(request):
_locals.request = request