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中的显式声明类型参数使用方式

《python中的显式声明类型参数使用方式》文章探讨了Python3.10+版本中类型注解的使用,指出FastAPI官方示例强调显式声明参数类型,通过|操作符替代Union/Optional,可提升代... 目录背景python函数显式声明的类型汇总基本类型集合类型Optional and Union(py

使用Python实现无损放大图片功能

《使用Python实现无损放大图片功能》本文介绍了如何使用Python的Pillow库进行无损图片放大,区分了JPEG和PNG格式在放大过程中的特点,并给出了示例代码,JPEG格式可能受压缩影响,需先... 目录一、什么是无损放大?二、实现方法步骤1:读取图片步骤2:无损放大图片步骤3:保存图片三、示php

Python文本相似度计算的方法大全

《Python文本相似度计算的方法大全》文本相似度是指两个文本在内容、结构或语义上的相近程度,通常用0到1之间的数值表示,0表示完全不同,1表示完全相同,本文将深入解析多种文本相似度计算方法,帮助您选... 目录前言什么是文本相似度?1. Levenshtein 距离(编辑距离)核心公式实现示例2. Jac

使用Python实现一个简易计算器的新手指南

《使用Python实现一个简易计算器的新手指南》计算器是编程入门的经典项目,它涵盖了变量、输入输出、条件判断等核心编程概念,通过这个小项目,可以快速掌握Python的基础语法,并为后续更复杂的项目打下... 目录准备工作基础概念解析分步实现计算器第一步:获取用户输入第二步:实现基本运算第三步:显示计算结果进

Python多线程实现大文件快速下载的代码实现

《Python多线程实现大文件快速下载的代码实现》在互联网时代,文件下载是日常操作之一,尤其是大文件,然而,网络条件不稳定或带宽有限时,下载速度会变得很慢,本文将介绍如何使用Python实现多线程下载... 目录引言一、多线程下载原理二、python实现多线程下载代码说明:三、实战案例四、注意事项五、总结引

Python利用PySpark和Kafka实现流处理引擎构建指南

《Python利用PySpark和Kafka实现流处理引擎构建指南》本文将深入解剖基于Python的实时处理黄金组合:Kafka(分布式消息队列)与PySpark(分布式计算引擎)的化学反应,并构建一... 目录引言:数据洪流时代的生存法则第一章 Kafka:数据世界的中央神经系统消息引擎核心设计哲学高吞吐

Python进阶之列表推导式的10个核心技巧

《Python进阶之列表推导式的10个核心技巧》在Python编程中,列表推导式(ListComprehension)是提升代码效率的瑞士军刀,本文将通过真实场景案例,揭示列表推导式的进阶用法,希望对... 目录一、基础语法重构:理解推导式的底层逻辑二、嵌套循环:破解多维数据处理难题三、条件表达式:实现分支

Java调用Python脚本实现HelloWorld的示例详解

《Java调用Python脚本实现HelloWorld的示例详解》作为程序员,我们经常会遇到需要在Java项目中调用Python脚本的场景,下面我们来看看如何从基础到进阶,一步步实现Java与Pyth... 目录一、环境准备二、基础调用:使用 Runtime.exec()2.1 实现步骤2.2 代码解析三、

Python中的filter() 函数的工作原理及应用技巧

《Python中的filter()函数的工作原理及应用技巧》Python的filter()函数用于筛选序列元素,返回迭代器,适合函数式编程,相比列表推导式,内存更优,尤其适用于大数据集,结合lamb... 目录前言一、基本概念基本语法二、使用方式1. 使用 lambda 函数2. 使用普通函数3. 使用 N

Python如何实现高效的文件/目录比较

《Python如何实现高效的文件/目录比较》在系统维护、数据同步或版本控制场景中,我们经常需要比较两个目录的差异,本文将分享一下如何用Python实现高效的文件/目录比较,并灵活处理排除规则,希望对大... 目录案例一:基础目录比较与排除实现案例二:高性能大文件比较案例三:跨平台路径处理案例四:可视化差异报