当前位置：首页 > news >正文

Django+Celery框架自动化定时任务开发

news 2025/7/10 18:20:00

本章介绍使用DjCelery即Django+Celery框架开发定时任务功能，在Autotestplat平台上实现单一接口自动化测试脚本、业务场景接口自动化测试脚本、App自动化测试脚本、Web自动化测试脚本等任务的定时执行、调度、管理等，从而取代Jenkins上的定时执行脚本和发送邮件等功能。**

自动化测试逻辑流程图11.1所示。

11.1 环境搭建

1．安装

步骤1 安装Celery。pyramid_celery-3.0.0，

配置https://pypi.python.org/pypi/pyramid_celery/。

步骤2 安装django-clery。django-celery-3.2.2，

配置https://pypi.python.org/pypi/django- celery。 INSTALLED_APPS= []

加入2：

'djcelery', 运行 Python manage.py migrate

步骤 3 安装celery-with-redis-3.0，

地址为https://pypi.python.org/pypi/celery-with-redis/。

步骤 4 安装django-clery-beat。django-celery-beat-1.1.0，

https://pypi.python.org/pypi/ django_celery_beat。

步骤5 下载Redis-x64-3.2.100，

Redis-x64-3.2.100.zip github.com/MicrosoftAr…

2．配置

步骤1 在Settings.py中增加如下内容。

加入1：

import djcelery

djcelery.setup_loader() #加载djcelery

加入2:

#数据库调度

CELERYBEAT_SCHEDULER ='djcelery.schedulers.DatabaseScheduler'

加入3:

BROKER_URL = 'redis://127.0.0.1:6379/0'

BROKER_TRANSPORT = 'redis'

步骤2 在应用Apitest目录下新建celery.py文件1，加入如下内容。

from future import absolute_import

import os

import django

from celery import Celery

from django.conf import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE','autotest.settings')

django.setup()

app = Celery('autotest')

app.config_from_object('django.conf:settings')

app.autodiscover_tasks(lambda:settings.INSTALLED_APPS)

步骤3 新建tasks.py文件，加入如下内容。

--coding:utf-8 --

importrequests, time, sys, re

importurllib, zlib#,

importpymysql

importunittest

from traceimport CoverageResults

importjson

fromidlelib.rpc import response_queue

fromapitest.celery import app

from timeimport sleep

@app.task

def hello_world():

print('已运行')

步骤4 启动服务python manage.py runserver。

步骤5 解压缩后，运行CMD，切换到相应目录，输入启动Redis指令redis-server redis. windows.conf，成功后出现如图11.2所示界面。

▲图11.2

步骤6 启动指令python manage.py celery worker -l info。

步骤7 启动指令python manage.py celery beat。

11.2 前端功能实现

1．功能描述

完成实现单一接口测试用例、业务场景接口API测试用例、AppUI测试用例、WebUI测试用例的自动化定时任务。

2．程序清单

在autotest\apitest\templates目录下新建periodic_task.html文件，加入如下内容。

<html>

<head>

{% load bootstrap4 %}

{% bootstrap_css %}

{% bootstrap_javascript %}

<title>产品自动化测试平台</title>

</head>

<div>

<ahref="#"> </a>

<ul>

</ul>

<ul>

</ul>

</div>

</nav>

<divstyle="padding-top: 70px;">

<formmethod="get" action="/tasksearch/">

{% csrf_token %}

</select>

<imgsrc="/static/admin/img/icon-changelink.svg" alt="修改"/>

</a>

<imgsrc="/static/admin/img/icon-addlink.svg" alt="增加"/>增加

</a>

</form>

</div>

<divstyle="padding-top: 20px;">

<div>

<thead>

<tr>

</tr>

</thead>

<tbody>

{% for task in tasks %}{% for periodic inperiodics %}

<tr>

{% if task.interval_id != null andtask.interval_id == periodic.id %}

<td><a style='color:green'>每{{ periodic.period }} {{ periodic.every}}次</a></td>

<td>{{ task.date_changed }}</td>

<td>{{ task.enabled }}</td>

<td>{% if task.id == 1 %}

{% elif task.id == 2 %}

{% else %}

{% endif %}

</td>

{% else %}

{% endif %}

{% for crontab in crontabs %}

{% if task.crontab_id != null and task.crontab_id ==crontab.id and task.interval_id == 1 %}

<td><a style='color:green'>{{crontab.month_of_year }}年{{crontab.day_of_month }}月{{crontab.day_of_week }}日{{crontab.hour }}时{{ crontab.minute}}分</a></td>

<td>{{ task.date_changed }}</td>

<td>{{ task.enabled }}</td>

{% else %}

{% endif %}

{% endfor %}{% endfor %}{% endfor %}

</tbody>

</table>

</div>

<span style="position:absolute;right:100px; bottom:30px;"> {# 把翻页功能固定显示在右下角#}

<tr><th>总数</th><td>{{ taskcounts }}</td></tr> {# 前端读取定义的变量#}

</div>

<div>

&lt;ulclass="pagination" id="pager"&gt;{#上一页链接开始#}{%if tasks.has_previous %}{#  如果有上一页，则正常显示上一页链接#}&lt;li&gt;&lt;ahref="/periodic_task/?page={{ tasks.previous_page_number }}"&gt;上一页&lt;/a&gt;&lt;/li&gt;    {#  上一页标签 #}{%else %}&lt;li class="previous disabled"&gt;&lt;ahref="#"&gt;上一页&lt;/a&gt;&lt;/li&gt;{# 如果当前不存在上一页，则上一页的链接不可单击#}{%endif %}{# 上一页链接开始#}{%for num in tasks.paginator.page_range %}{% if num == currentPage %}&lt;li&gt;&lt;a href="/periodic_task/?page={{ num }}"&gt;{{ num}}&lt;/a&gt;&lt;/li&gt; {#显示当前页数链接#}{% else %}&lt;liclass="item"&gt;&lt;a href="/periodic_task/?page={{ num}}"&gt;{{ num }}&lt;/a&gt;&lt;/li&gt;{% endif %}{% endfor %}{# 下一页链接开始#}{% if tasks.has_next %} {#  如果有下一页，则正常显示下一页链接#}&lt;liclass="next"&gt;&lt;a href="/periodic_task/?page={{tasks.next_page_number }}"&gt;下一页&lt;/a&gt;&lt;/li&gt;{% else %}&lt;li&gt;&lt;a href="#"&gt;下一页&lt;/a&gt;&lt;/li&gt;{% endif %}{# 下一页链接结束#}&lt;/ul&gt;

</div>

</body>

</html>

功能描述：实现自动化测试任务调度执行，包括单一接口、扫描、流程接口、业务场景、Web搜索、自动化平台测试开发、App登录，CSDN定时任务注册，定时任务执行等功能。

程序清单：在apitest/views.py中加入如下内容。

from .tasks importhello_world

from .tasks importtest_readSQLcase

from djcelery.modelsimport PeriodicTask,CrontabSchedule,IntervalSchedule

任务计划

@login_required

defperiodic_task(request):

username = request.session.get('user', '')task_list = PeriodicTask.objects.all()task_count =PeriodicTask.objects.all().count()  #统计数periodic_list =IntervalSchedule.objects.all()  # 周期任务（如每隔1小时执行1次）crontab_list =CrontabSchedule.objects.all()    # 定时任务（如某年某月某日的某时，每# 天的某时）paginator = Paginator(task_list, 5)  #生成paginator对象,设置每页显示5条记录page = request.GET.get('page',1)  #获取当前的页码数,默认为第1页currentPage=int(page)  #把获取的当前页码数转换成整数类型try:task_list = paginator.page(page)#获取当前页码数的记录列表except PageNotAnInteger:task_list = paginator.page(1)#如果输入的页数不是整数，则显示第1页内容except EmptyPage:task_list =paginator.page(paginator.num_pages)#如果输入的页数不在系统的页数中，# 则显示最后一页内容return render(request,"periodic_task.html", {"user": username,"tasks":task_list,"taskcounts": task_count, "periodics":periodic_list,"crontabs": crontab_list })

搜索功能

@login_required

deftasksearch(request):

username = request.session.get('user', '')# 读取浏览器登录Sessionsearch_name =request.GET.get("task", "")task_list = PeriodicTask.objects.filter(task__icontains=search_name)periodic_list =IntervalSchedule.objects.all()  # 周期任务（如每隔1小时执行1次）crontab_list =CrontabSchedule.objects.all()    # 定时任务（如某年某月某日的某时，每# 天的某时）return render(request,'periodic_task.html',{"user": username,"tasks":task_list,"periodics":periodic_list,"crontabs": crontab_list })

在autotest/urls.py中加入：

path('periodic_task/',views.periodic_task),

path('tasksearch/', views.tasksearch),

在apitest/left.html中加入：

                &lt;li&gt;&lt;a  href="../periodic_task"target="mainFrame"&gt;&lt;iclass="glyphicon glyphicon-fire"&gt;&lt;/i&gt;任务计划       &lt;/a&gt;&lt;/li&gt;

<tr><td>&nbsp;</td></tr>

查看前端页面效果，如图11.3所示。

▲图11.3

行动吧，在路上总比一直观望的要好，未来的你肯定会感谢现在拼搏的自己！如果想学习提升找不到资料，没人答疑解惑时，请及时加入扣群： 320231853，里面有各种软件测试+开发资料和技术可以一起交流学习哦。

最后感谢每一个认真阅读我文章的人，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走：

这些资料，对于【软件测试】的朋友来说应该是最全面最完整的备战仓库，这个仓库也陪伴上万个测试工程师们走过最艰难的路程，希望也能帮助到你！

Django+Celery框架自动化定时任务开发

--coding:utf-8 --

任务计划

搜索功能

相关文章：

Django+Celery框架自动化定时任务开发

解决element-plus table组件 fixed=“right“(left)浮动后横向滚动文字穿透的问题

【opencv】示例-distrans.cpp 距离变换

LVGL V8 代码细读——极致的链表使用

蓝桥杯第十二届c++大学B组详解

Tubi 十岁啦！

Qt C++ 实现文件监视源码

蓝桥杯第十一届c++大学B组详解

大模型日报2024-04-10

redis修改协议改了，有哪些替代品？

《QT实用小工具·十六》IP地址输入框控件

windows 系统下 mysql 数据库的下载与安装（包括升级安装）

Redis Stack十部曲之三：理解Redis Stack中的数据类型

OneForAll安装使用

【现代C++】线程支持库

游戏引擎架构01__引擎架构图

[Java、Android面试]_15_Android为什么使用Binder？

Python+Selenium+Unittest 之Unittest3(TestSuite()和TextTestRunner())

3D桌面端可视化引擎HOOPS Visualize如何实现3D应用快速开发？

Vue探索之Vue2.x源码分析（二）

网络编程（Modbus进阶）

反向工程与模型迁移：打造未来商品详情API的可持续创新体系

java 实现excel文件转pdf | 无水印 | 无限制

《Playwright：微软的自动化测试工具详解》

可靠性+灵活性：电力载波技术在楼宇自控中的核心价值

【RockeMQ】第2节｜RocketMQ快速实战以及核⼼概念详解（二）

[Java恶补day16] 238.除自身以外数组的乘积

在WSL2的Ubuntu镜像中安装Docker

AI书签管理工具开发全记录（十九）：嵌入资源处理

Docker 本地安装 mysql 数据库