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

目录
  • 使用背景
  • 什么是 APScheduler 框架？
  • APScheduler 框架包含四个组成部分
  • APScheduler 在 flask 中使用
    • 编写任务函数，开始 APScheduler 的调度
    • 部分项目代码
  • 总结

使用背景

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

什么是 APScheduler 框架？

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

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

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的应用示例的详细内容，更多关于python自动化测试APScheduler Flask的资料请关注我们其它相关文章！

赞 (0)

相关推荐

  • Python + selenium 自动化测试框架详解

    Python + selenium 自动化测试框架详解

    目录 自动化测试框架 1.基础层(通用层) 2.功能层(页面层) 3.业务层 4.用例层 4层框架对应的项目工程 总结 自动化测试框架 项目自动化测试框架设计为4层 1.基础层(通用层) 基础层: 将通用,重复性比较高的代码封装到这里. 写通用的代码的. 其他3层想要的话,就可以直接调用. 例如: 读取测试数据的代码,读取配置信息的代码:截图的代码的,定位元素的代码等等 2.功能层(页面层) 功能层: pages 封装页面的. 把页面封装成类.类中包含:属性和方法 把页面上的界面元素定义成类的属

  • 详解Python下Flask-ApScheduler快速指南

    引言:Flask是Python社区非常流行的一个Web开发框架,本文将尝试将介绍APScheduler应用于Flask之中. 1. Flask介绍 Flask是Python社区大名鼎鼎的"microframework",基于简单的核心,使用extension来增加其他功能,其提供非常丰富易用的扩展包, 比如: 2.  Flask-APScheduler 社区提供了一个Flask-APScheduler的模块,方便大家直接在Flask模块中使用APScheduler. 关于安装的命令,仍

  • python+appium自动化测试之如何控制App的启动和退出

    python+appium自动化测试之如何控制App的启动和退出

    目录 一.启动app 二.退出App 三.appium实现原理 四.如何获取deviceName? 五.如何获取apk的appActivity和appPackage? 由于本人使用的是Android设备做自动化测试,所以以下内容均基于Android系统做出的整理 一.启动app 启动app需要设置Capability参数,而Capability参数放在Desired Capalibity中,Desired Capalibity告诉Appium想要的自动化平台和应用程序,这是一组键值对,主要是用于

  • flask 使用 flask_apscheduler 做定时循环任务的实现

    我是初学者,对 flask 很陌生,网上搜到的文章都看不懂,很尴尬. 本意是打算对广发信用卡diy卡积分兑换签帐额的数量进行爬虫监控.将抓取到的余量通过钉钉机器人发送到群里.爬虫代码就不写了,这里只写定时任务每分钟抓一次的代码. 不扯淡,直接干 以下代码,都是当前文件和定时器相关的代码,没有将其它代码写进来辣各位的眼睛. 文件夹结构图: ├── app │ ├── test.py │ ├── __init__.py │ └── routes.py ├── config.py ├── requir

  • Appium+Python+pytest自动化测试框架的实战

    Appium+Python+pytest自动化测试框架的实战

    菜鸟一枚,写的不好勿喷,大家一起学习 先简单介绍一下目录,再贴一些代码,代码里有注释 Basic目录下写的是一些公共的方法,Data目录下写的是测试数据,image存的是测试失败截图,Log日志文件,Page测试的定位元素,report测试报告,Test测试用例,pytest.ini是pytest启动配置文件,requirements.txt需要安装的py模块,run.py运行文件 Basic/base.py 里面封装了 一些方法,元素的点击,输入,查找,还有一些自己需要的公共方法也封装在里面,

  • python+appium实现自动化测试的示例代码

    python+appium实现自动化测试的示例代码

    目录 1.什么是Appium 2.启动一个app自动化程序的步骤 3.appium服务介绍 4. appium客户端使用 5.adb的使用 6.Appium启动过程分析 1.什么是Appium appium是一个开源的测试自动化框架,可以与原生的.混合的和移动的web应用程序一直使用.它使用WebDriver协议驱动IOS(内置的测试引擎xcuitest).Android(uiautomator2,Espresso)和Windows应用程序 原生应用程序:安卓程序是用JAVA或kotlin开发出

  • python Airtest自动化测试工具的的使用

    python Airtest自动化测试工具的的使用

    目录 Airtest的官网介绍项目构成 如何上手? 一.安装 二.使用 三.在本地环境中使用及遇到的问题 1.使用"pip install airtest" 进行安装 2.IDE连接不存在设备或被占用设备容易假死 3. 假死期间关闭IDE会造成adb驻留 4.尝试连接不存在设备时,即使设备是不存在的也会加入列表 四.最后啰嗦几句 Airtest全称AirtestProject,是由网易游戏推出的一款自动化测试框架,在软件测试的时候使用到了该框架.这里记录一下安装.使用和遇到的坑的问题…

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

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

    目录 使用背景 什么是 APScheduler 框架? APScheduler 框架包含四个组成部分 APScheduler 在 flask 中使用 编写任务函数,开始 APScheduler 的调度 部分项目代码 总结 使用背景 实际项目中,需要验证打点数据在各个系统中收集是否一致,而部分节点打点数据收集是通过异步任务实现的,等待时间比较久.为应对业务异步操作处理,实现异步数据的收集,经过调研后,选择了 APScheduler 框架. 什么是 APScheduler 框架? APSchedul

  • python自动化测试selenium定位frame及iframe示例

    python自动化测试selenium定位frame及iframe示例

    frame标签有frameset.frame.iframe三种,frameset和其它普通标签没有区别,不会影响正常定位,而frame与iframe对selenium定位而言是一样的. Selenium有以下方法对frame进行操作. 示例网站:http://sahitest.com/demo/framesTest.htm 示例脚本: from selenium import webdriver from time import sleep class TestFrame(object): de

  • Python自动化测试中yaml文件读取操作

    什么是yaml 一种标记语言.yaml 是专门用来写配置文件的语言,非常简洁和强大 更直观,更方便,有点类似于json格式 yaml文件格式:test.yaml 安装yaml pip install pyyaml yaml基本语法规则 大小写敏感 使用缩进表示层级关系 缩进时不允许使用Tab键,只允许使用空格. 缩进的空格数目不重要,只要相同层级的元素左侧对齐即可 #表示注释,从这个字符一直到行尾,都会被解析器忽略,这个和python的注释一样 键值对(dict) yaml文件 user: ad

  • python自动化测试selenium执行js脚本实现示例

    WebDriver有2个方法执行Java Script脚本. (1)同步执行:execute_script (2)异步执行:execute_async_script from selenium import webdriver from time import sleep class TestScriptss(object): def setup(self): self.driver = webdriver.Chrome() self.driver.get(http://www.baidu.co

  • python自动化测试中装饰器@ddt与@data源码深入解析

    目录 一.使用ddt和data装饰器的大致框架如下,每个test_开头的方法,代表一条测试用例 二.给类动态的增加方法 案例1 案例2: 案例3: 三.ddt和data的源码解析 原因: 解决: 分部解析代码 总结 一.使用ddt和data装饰器的大致框架如下,每个test_开头的方法,代表一条测试用例 from ddt import ddt,data import unittest test_datas=[ {'id':1,'title':'测试用例1'}, {'id':2,'title':'

  • Python PyQt5中窗口数据传递的示例详解

    目录 单一窗口数据传递 多窗口数据传递:调用属性 多窗口数据传递:信号与槽 开发应用程序时,若只有一个窗口则只需关心这个窗口里面的各控件之间如何传递数据.如果程序有多个窗口,就要关心不同的窗口之间是如何传递数据. 单一窗口数据传递 对于单一窗口的程序来说,一个控件的变化会影响另一个控件的变化通过信号与槽的机制就可简单解决. import sys from PyQt5.QtWidgets import QWidget, QLCDNumber, QSlider, QVBoxLayout, QAppl

  • Python编程中装饰器的使用示例解析

    装饰函数和方法 我们先定义两个简单的数学函数,一个用来计算平方和,一个用来计算平方差: # get square sum def square_sum(a, b): return a**2 + b**2 # get square diff def square_diff(a, b): return a**2 - b**2 print(square_sum(3, 4)) print(square_diff(3, 4)) 在拥有了基本的数学功能之后,我们可能想为函数增加其它的功能,比如打印输入.我们

  • Python程序中的观察者模式结构编写示例

    Python程序中的观察者模式结构编写示例

    察者模式定义 定义了对象之间的一对多依赖,这样一来,当一个对象改变状态时,它的所有依赖都会收到通知并自动更新.观察者模式提供了一种对象设计,让主题和观察者之间松耦合. 设计原则 为了交互对象之间的松耦合设计而努力.松耦合的设计之所以能让我们建立有弹性的系统,能够应对变化,是因为对象之间的互相依赖降到了最低. 模式结构与说明 1.Subject:主题(目标)接口,主题可以注册.移除对其感兴趣的观察者,在自身数据发生变化的时候,还可以通知这些观察者.一个主题可以有多个观察者,需要注意的是,观察者的顺

  • python案例中Flask全局配置示例详解

    目录 WEB服务全局配置 Flask全局配置 before_request after_request Flask自定义中间件 WEB服务全局配置 在目前的开发过市场当中,有很多WEB服务框架,Flask只是其中之一,但是总体上来看,所有的WEB框架都是依据HTTP协议的逻辑从请求到响应设计的.固然有很多功能是独立的,但是也有一部分功能需要全局设定,比如安全校验,比如埋点日志,那么这里就用到了全局配置. 所谓的全局配置,就是在框架全局,请求前后,响应前后,设置的全局配置,比如登录校验,这个功能并

  • Python定时库APScheduler的原理以及用法示例

    Python定时库APScheduler的原理以及用法示例

    目录 1. APScheduler简介 2. APScheduler组件 2.1. APScheduler中几个重要的概念 2.1.1. Job 作业 2.1.2. Trigger 触发器 2.1.3. Executor 执行器 2.1.4. Jobstore 作业存储 2.1.5. Event 事件 2.1.6. Listener 监听事件 2.1.7. Scheduler 调度器 2.2. Scheduler工作流程图 2.2.1. Scheduler添加job流程 2.2.2 Schedu

随机推荐