python自動化測試中APScheduler?Flask怎么應用

這篇文章主要介紹“python自動化測試中APScheduler Flask怎么應用”，在日常操作中，相信很多人在python自動化測試中APScheduler Flask怎么應用問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”python自動化測試中APScheduler Flask怎么應用”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

使用背景

實際項目中，需要驗證打點數據在各個系統中收集是否一致，而部分節點打點數據收集是通過異步任務實現的，等待時間比較久。為應對業務異步操作處理，實現異步數據的收集，經過調研后，選擇了 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 函數可以指定停止條件，在本項目中，因為步驟比較多，一旦有環節出錯，就需要結束任務，保存已驗證的環節，因此在拿到任務結果時，不論是整個驗證成功的結果，還是某個環節出錯的結果，都會停止本次調度，結束掉本次驗證。

到此，關于“python自動化測試中APScheduler Flask怎么應用”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！