Python自动化测试中APScheduler Flask的应用示例

2024-01-20 00:04

本文主要是介绍Python自动化测试中APScheduler Flask的应用示例,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

使用背景
实际项目中,需要验证打点数据在各个系统中收集是否一致,而部分节点打点数据收集是通过异步任务实现的,等待时间比较久。为应对业务异步操作处理,实现异步数据的收集,经过调研后,选择了 APScheduler 框架。


什么是 APScheduler 框架?
APScheduler 是基于 Quartz(一个功能丰富的开源任务调度系统) 的一个 Python 定时任务框架,使用起来简单且方便,提供了基于日期、固定时间间隔以及 crontab 类型的任务,并且可以持久化任务,基于这些功能可以快速实现 python 的定时轮询任务系统。

使用 APScheduler 框架,可以通过 pip 安装

1

pip install apscheduler

APScheduler 框架包含四个组成部分

触发器 (trigger)触发器包含调度逻辑,每一个作业有它自己的触发器,用于决定接下来哪一个作业会运行;除了他们自己初始的配置,触发器本身是无状态的。

作业存储 (job store)作业存储存储被调度的作业,默认的作业存储是简单的将作业保存到内存中,如果选择其他方式也可以将作业保存到数据库中;一个作业数据的保存将会在持久化作业存储的时候被序列化,然后在加载时被反序列化;调度器无法分享同一个作业存储。

执行器 (executor)执行器处理作业的运行,一般通过在作业中提交制定好的可调用对象到一个线程中或者线程池中来执行;在作业完成时,执行器会去通知调度器。

调度器 (scheduler)调度器是 APScheduler 的核心,所有相关的组件都要通过它来定义,已配置好的任务也是要通过它来调度。


APScheduler 在 flask 中使用
因为 scheduler 任务需要耗费较多时间,因此当在项目中收到 flask 的接口请求后,可以通过线程异步处理耗时任务,先将 “正在处理” 作为接口结果返回,
示例代码结构如下:

编写任务函数,开始 APScheduler 的调度

在通过 flask 接口拿到需要的任务参数后,便可以创建调度任务。在创建调度任务之前,我们需要确定要选择哪一种调度器、job 存储、执行器和触发器,
调度器的选择主要基于编程环境以及 APScheduler 的用途,

这里我们根据需要选择 BackgroundScheduler。

在 job 存储的选择上,需要根据自己的 job 是否需要持久化,因为没有特殊的需求,所以使用默认的内存方式

执行器的选择需要依据 job 的类型,默认的线程池执行器apscheduler.executors.pool.ThreadPoolExecutor 已经可以满足大多数情况。

管理 job 的调度方式需要选择一个合适的触发器,APScheduler 内置三种触发器;

因为我们的自动化需要对各个子环节进行验证,当上一个环节成功后才能进行下一个环节的验证,因此选择 apscheduler.triggers.interval,以固定的时间间隔运行 job。

部分项目代码

 periodic_task 是项目中的任务调度函数;首先实例化了一个 BackgroundScheduler 调度器,接着向调度器添加 job,添加的 job 为 data_task 函数,同时定义了 job 的触发器,指定固定的时间间隔为 58 秒。

其中 data_task 描述了具体的 job 细节,即分别判断当前不同的任务节点执行相应的验证过程,并将每一步的验证状态记录到数据库中,这样在下一次执行 data_task 时,就可以去校验新的环节;

启动调度器使用 start 函数,结束调度器使用 shutdown 函数;

shutdown 函数可以指定停止条件,在本项目中,因为步骤比较多,一旦有环节出错,就需要结束任务,保存已验证的环节,因此在拿到任务结果时,不论是整个验证成功的结果,还是某个环节出错的结果,都会停止本次调度,结束掉本次验证。


总结
APScheduler 在 flask 中使用需要用到线程池异步去处理耗时任务;
使用 APScheduler 需要配置好合适的调度器、job 存储、执行器和触发器;
在业务中验证复杂连续的步骤可以使用轮询的方式,并设置好任务结束的条件,不仅可以校验每一步的验证结果而且有环节出错也不影响整个流程。
最后感谢每一个认真阅读我文章的人,礼尚往来总是要有的,虽然不是什么很值钱的东西,如果你用得到的话可以直接拿走:

文档获取方式:点击右边链接领取:软件测试全套资料分享         

这篇关于Python自动化测试中APScheduler Flask的应用示例的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/624165

相关文章

Python将博客内容html导出为Markdown格式

《Python将博客内容html导出为Markdown格式》Python将博客内容html导出为Markdown格式,通过博客url地址抓取文章,分析并提取出文章标题和内容,将内容构建成html,再转... 目录一、为什么要搞?二、准备如何搞?三、说搞咱就搞!抓取文章提取内容构建html转存markdown

Python获取中国节假日数据记录入JSON文件

《Python获取中国节假日数据记录入JSON文件》项目系统内置的日历应用为了提升用户体验,特别设置了在调休日期显示“休”的UI图标功能,那么问题是这些调休数据从哪里来呢?我尝试一种更为智能的方法:P... 目录节假日数据获取存入jsON文件节假日数据读取封装完整代码项目系统内置的日历应用为了提升用户体验,

Python FastAPI+Celery+RabbitMQ实现分布式图片水印处理系统

《PythonFastAPI+Celery+RabbitMQ实现分布式图片水印处理系统》这篇文章主要为大家详细介绍了PythonFastAPI如何结合Celery以及RabbitMQ实现简单的分布式... 实现思路FastAPI 服务器Celery 任务队列RabbitMQ 作为消息代理定时任务处理完整

Python Websockets库的使用指南

《PythonWebsockets库的使用指南》pythonwebsockets库是一个用于创建WebSocket服务器和客户端的Python库,它提供了一种简单的方式来实现实时通信,支持异步和同步... 目录一、WebSocket 简介二、python 的 websockets 库安装三、完整代码示例1.

揭秘Python Socket网络编程的7种硬核用法

《揭秘PythonSocket网络编程的7种硬核用法》Socket不仅能做聊天室,还能干一大堆硬核操作,这篇文章就带大家看看Python网络编程的7种超实用玩法,感兴趣的小伙伴可以跟随小编一起... 目录1.端口扫描器:探测开放端口2.简易 HTTP 服务器:10 秒搭个网页3.局域网游戏:多人联机对战4.

使用Python实现快速搭建本地HTTP服务器

《使用Python实现快速搭建本地HTTP服务器》:本文主要介绍如何使用Python快速搭建本地HTTP服务器,轻松实现一键HTTP文件共享,同时结合二维码技术,让访问更简单,感兴趣的小伙伴可以了... 目录1. 概述2. 快速搭建 HTTP 文件共享服务2.1 核心思路2.2 代码实现2.3 代码解读3.

C#使用SQLite进行大数据量高效处理的代码示例

《C#使用SQLite进行大数据量高效处理的代码示例》在软件开发中,高效处理大数据量是一个常见且具有挑战性的任务,SQLite因其零配置、嵌入式、跨平台的特性,成为许多开发者的首选数据库,本文将深入探... 目录前言准备工作数据实体核心技术批量插入:从乌龟到猎豹的蜕变分页查询:加载百万数据异步处理:拒绝界面

Python使用自带的base64库进行base64编码和解码

《Python使用自带的base64库进行base64编码和解码》在Python中,处理数据的编码和解码是数据传输和存储中非常普遍的需求,其中,Base64是一种常用的编码方案,本文我将详细介绍如何使... 目录引言使用python的base64库进行编码和解码编码函数解码函数Base64编码的应用场景注意

用js控制视频播放进度基本示例代码

《用js控制视频播放进度基本示例代码》写前端的时候,很多的时候是需要支持要网页视频播放的功能,下面这篇文章主要给大家介绍了关于用js控制视频播放进度的相关资料,文中通过代码介绍的非常详细,需要的朋友可... 目录前言html部分:JavaScript部分:注意:总结前言在javascript中控制视频播放

Python基于wxPython和FFmpeg开发一个视频标签工具

《Python基于wxPython和FFmpeg开发一个视频标签工具》在当今数字媒体时代,视频内容的管理和标记变得越来越重要,无论是研究人员需要对实验视频进行时间点标记,还是个人用户希望对家庭视频进行... 目录引言1. 应用概述2. 技术栈分析2.1 核心库和模块2.2 wxpython作为GUI选择的优