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 函數可以指定停止條件，在本專案中，因為步驟比較多，一旦有環節出錯，就需要結束任務，儲存已驗證的環節，因此在拿到任務結果時，不論是整個驗證成功的結果，還是某個環節出錯的結果，都會停止本次排程，結束掉本次驗證。

總結

• APScheduler 在 flask 中使用需要用到執行緒池非同步去處理耗時任務；
• 使用 APScheduler 需要設定好合適的排程器、job 儲存、執行器和觸發器；
• 在業務中驗證複雜連續的步驟可以使用輪詢的方式，並設定好任務結束的條件，不僅可以校驗每一步的驗證結果而且有環節出錯也不影響整個流程。

以上就是python自動化測試中APScheduler Flask的應用範例的詳細內容，更多關於python自動化測試APScheduler Flask的資料請關注it145.com其它相關文章！