python自動化測試中APScheduler?Flask得應用示例

    目錄

    使用背景

    實際項目中,需要驗證打點數據在各個系統中收集是否一致,而部分節(jié)點打點數據收集是通過異步任務實現得,等待時間比較久。為應對業(yè)務異步操作處理,實現異步數據得收集,經過調研后,選擇了 APScheduler 框架。

    什么是 APScheduler 框架?

    APScheduler 是基于 Quartz(一個功能豐富得開源任務調度系統) 得一個 Python 定時任務框架,使用起來簡單且方便,提供了基于日期、固定時間間隔以及 crontab 類型得任務,并且可以持久化任務,基于這些功能可以快速實現 python 得定時輪詢任務系統。

    使用 APScheduler 框架,可以通過 pip 安裝

    pip install apscheduler

    APScheduler 框架包含四個組成部分

    觸發(fā)器 (trigger)觸發(fā)器包含調度邏輯,每一個作業(yè)有它自己得觸發(fā)器,用于決定接下來哪一個作業(yè)會運行;除了他們自己初始得配置,觸發(fā)器本身是無狀態(tài)得。

    作業(yè)存儲 (job store)作業(yè)存儲存儲被調度得作業(yè),默認得作業(yè)存儲是簡單得將作業(yè)保存到內存中,如果選擇其他方式也可以將作業(yè)保存到數據庫中;一個作業(yè)數據得保存將會在持久化作業(yè)存儲得時候被序列化,然后在加載時被反序列化;調度器無法分享同一個作業(yè)存儲。

    執(zhí)行器 (executor)執(zhí)行器處理作業(yè)得運行,一般通過在作業(yè)中提交制定好得可調用對象到一個線程中或者線程池中來執(zhí)行;在作業(yè)完成時,執(zhí)行器會去通知調度器。

    調度器 (scheduler)調度器是 APScheduler 得核心,所有相關得組件都要通過它來定義,已配置好得任務也是要通過它來調度。

    APScheduler 在 flask 中使用

    因為 scheduler 任務需要耗費較多時間,因此當在項目中收到 flask 得接口請求后,可以通過線程異步處理耗時任務,先將 “正在處理” 作為接口結果返回,
    示例代碼結構如下:

    編寫任務函數,開始 APScheduler 得調度

    在通過 flask 接口拿到需要得任務參數后,便可以創(chuàng)建調度任務。在創(chuàng)建調度任務之前,我們需要確定要選擇哪一種調度器、job 存儲、執(zhí)行器和觸發(fā)器,
    調度器得選擇主要基于編程環(huán)境以及 APScheduler 得用途,

    這里我們根據需要選擇 BackgroundScheduler。

    在 job 存儲得選擇上,需要根據自己得 job 是否需要持久化,因為沒有特殊得需求,所以使用默認得內存方式

    執(zhí)行器得選擇需要依據 job 得類型,默認得線程池執(zhí)行器apscheduler.executors.pool.ThreadPoolExecutor 已經可以滿足大多數情況。

    管理 job 得調度方式需要選擇一個合適得觸發(fā)器,APScheduler 內置三種觸發(fā)器;

    因為我們得自動化需要對各個子環(huán)節(jié)進行驗證,當上一個環(huán)節(jié)成功后才能進行下一個環(huán)節(jié)得驗證,因此選擇 apscheduler.triggers.interval,以固定得時間間隔運行 job。

    部分項目代碼

    periodic_task 是項目中得任務調度函數;首先實例化了一個 BackgroundScheduler 調度器,接著向調度器添加 job,添加得 job 為 data_task 函數,同時定義了 job 得觸發(fā)器,指定固定得時間間隔為 58 秒。

    其中 data_task 描述了具體得 job 細節(jié),即分別判斷當前不同得任務節(jié)點執(zhí)行相應得驗證過程,并將每一步得驗證狀態(tài)記錄到數據庫中,這樣在下一次執(zhí)行 data_task 時,就可以去校驗新得環(huán)節(jié);

    啟動調度器使用 start 函數,結束調度器使用 shutdown 函數;

    shutdown 函數可以指定停止條件,在本項目中,因為步驟比較多,一旦有環(huán)節(jié)出錯,就需要結束任務,保存已驗證得環(huán)節(jié),因此在拿到任務結果時,不論是整個驗證成功得結果,還是某個環(huán)節(jié)出錯得結果,都會停止本次調度,結束掉本次驗證。

    總結

    • APScheduler 在 flask 中使用需要用到線程池異步去處理耗時任務;
    • 使用 APScheduler 需要配置好合適得調度器、job 存儲、執(zhí)行器和觸發(fā)器;
    • 在業(yè)務中驗證復雜連續(xù)得步驟可以使用輪詢得方式,并設置好任務結束得條件,不僅可以校驗每一步得驗證結果而且有環(huán)節(jié)出錯也不影響整個流程。

    以上就是python自動化測試中APScheduler Flask得應用示例得詳細內容,更多關于python自動化測試APScheduler Flask得資料請關注之家其它相關內容!

    聲明:所有內容來自互聯網搜索結果,不保證100%準確性,僅供參考。如若本站內容侵犯了原著者的合法權益,可聯系我們進行處理。
    發(fā)表評論
    更多 網友評論1 條評論)
    暫無評論

    返回頂部

    主站蜘蛛池模板: 3d动漫精品啪啪一区二区中文 | 无码人妻精品一区二| 国产在线一区二区三区在线| 中文字幕亚洲一区二区va在线| 日韩精品一区二区三区中文3d| 色狠狠一区二区三区香蕉| 香蕉免费看一区二区三区| 国产在线一区观看| 国产精品资源一区二区| 亚洲av无码一区二区三区四区| 亚洲国产精品自在线一区二区| 成人区精品一区二区不卡| 精品一区二区三区在线视频观看 | 一色一伦一区二区三区| 久久国产精品最新一区| 国产亚洲福利精品一区| 日韩AV无码一区二区三区不卡毛片 | 精品无码日韩一区二区三区不卡| 国偷自产Av一区二区三区吞精| 色精品一区二区三区| 精品视频午夜一区二区| 国产精品va一区二区三区| 日本精品一区二区三区视频 | 亚洲狠狠久久综合一区77777| 国产一区二区中文字幕| 看电影来5566一区.二区| 高清一区二区三区日本久| 熟女精品视频一区二区三区| 亚洲一区电影在线观看| 亚洲无线码一区二区三区| 一区国严二区亚洲三区| 国产福利电影一区二区三区,免费久久久久久久精| 亚洲AV一区二区三区四区| 亚洲欧美成人一区二区三区| 国产一区二区在线| 精品免费AV一区二区三区| 中文字幕在线视频一区| 无码日韩精品一区二区三区免费 | 国产在线不卡一区二区三区| 国产av天堂一区二区三区| 老熟女高潮一区二区三区|