pypyr流水線YAML語法完全解析步驟、步驟組與成功/失敗回調(diào)一次看懂【免費下載鏈接】pypyrpypyr task-runner cli api for automation pipelines. Automate anything by combining commands, different scripts in different languages applications into one pipeline process.項目地址: https://gitcode.com/gh_mirrors/py/pypyrpypyr 是一款用 YAML 定義自動化任務運行器task-runner的開源工具它把命令、腳本和應用編排成一條 pypyr 流水線pipeline順序執(zhí)行。本文完整解析 pypyr 流水線的 YAML 語法每個步驟step支持哪些字段、如何用步驟組step group拆分可復用流程以及on_success/on_failure兩大回調(diào)如何兜底——一篇看懂直接上手。為什么用 YAML 寫自動化pypyr 的定位傳統(tǒng) shell 腳本或 Makefile 寫多了都知道頭疼換平臺就崩、變量傳遞混亂、錯誤處理靠硬扛。pypyr 的思路很直接——用一個 YAML 文件描述整條自動化流程然后一條命令跑完$ pip install --upgrade pypyr $ pypyr my-pipeline # 文件名不帶 .yaml? 跨平臺Linux / macOS / Windows 通吃CI 服務器和容器里同樣能跑? 多語言混編同一條流水線里可以跑 shell 命令、Python 代碼、調(diào)外部程序? 流程可控循環(huán)、條件、重試、失敗兜底全部在 YAML 里聲明想深入了解項目結(jié)構(gòu)可從 pypyr/pipeline.py 開始它定義了一次流水線運行的運行時對象。流水線文件的最小骨架一個 pypyr 流水線 YAML 的最小形態(tài)只需要一個steps鍵# my-pipeline.yaml steps: - name: pypyr.steps.echo in: echoMe: 你好pypyr 流水線執(zhí)行pypyr my-pipeline即可看到輸出。如果想讓流水線接收命令行參數(shù)加上可選的context_parsercontext_parser: pypyr.parser.keyvaluepairs steps: - name: pypyr.steps.echo in: echoMe: hello {who}隨后pypyr my-pipeline whoworldYAML 里的{who}就會被自動替換成world——這就是 pypyr 的變量替換變量來自 context 上下文。步驟的兩種寫法簡單與復雜簡單步驟一行搞定步驟名就是步驟模塊的完整名稱沒有參數(shù)時直接寫一行steps: - pypyr.steps.echo - pypyr.steps.now復雜步驟name in 參數(shù)需要傳參時用字典形式核心就兩個字段steps: - name: pypyr.steps.cmd # 要執(zhí)行的步驟模塊 in: # 注入步驟的輸入?yún)?shù) cmd: echo 任意命令in參數(shù)的生命周期在源碼里寫得很清楚執(zhí)行前注入 context執(zhí)行完自動移除不會污染后續(xù)步驟見 pypyr/dsl.py。內(nèi)置步驟如pypyr.steps.echo、pypyr.steps.cmd、pypyr.steps.py都位于 pypyr/steps/ 目錄每個文件名就是一個步驟。步驟裝飾器速查表一次看懂 7 個字段復雜步驟除了name和in還支持這些裝飾器字段解析邏輯見 pypyr/dsl.py 的_init_from_dict字段默認值作用runTrue表達式為真才執(zhí)行支持變量替換skipFalse表達式為真則跳過本步驟swallowFalse出錯時吞掉異常流水線繼續(xù)走onError無出錯時附加的自定義錯誤信息description無運行時在控制臺打印的步驟說明foreach無對列表逐項循環(huán)執(zhí)行retry無失敗自動重試可配退避策略while無條件成立則反復循環(huán)執(zhí)行條件執(zhí)行是最常用的場景例如steps: - name: pypyr.steps.cmd run: {deployEnabled} # 只有 deployEnabled 為 True 才執(zhí)行 in: cmd: echo 執(zhí)行部署容錯場景用swallow某步驟失敗不想中斷整條流水線時- name: pypyr.steps.cmd swallow: True in: cmd: 可能失敗的命令循環(huán)與重試讓流水線自己跑foreach 循環(huán)列表里每項執(zhí)行一遍當前項通過{i}引用steps: - name: pypyr.steps.cmd foreach: [once, twice, thrice] in: cmd: echo say {i}while 循環(huán)條件滿足就重復跑內(nèi)置{whileCounter}記錄輪次max限制最多輪數(shù)steps: - name: pypyr.steps.cmd while: max: 3 in: cmd: echo 第 {whileCounter} 輪retry 重試失敗后按退避策略fixed / linear / jitter重試pypyr/retries.py提供退避算法配置支持max、sleep、retryOn、stopOn等字段解析邏輯見 pypyr/dsl.py。步驟組把流水線切成可復用的塊這是 pypyr YAML 語法里最容易被忽略、也最有價值的設計流水線 YAML 的每個頂層鍵都是一個步驟組steps只是默認名。steps: - name: pypyr.steps.echo in: echoMe: 主流程 - name: pypyr.steps.call in: call: groups: [sg1] # 調(diào)用同文件里的步驟組 sg1 sg1: # 自定義步驟組 - name: pypyr.steps.echo in: echoMe: 我是被調(diào)用的 sg1steps組默認最先執(zhí)行pypyr.steps.call可跳轉(zhuǎn)調(diào)用其他命名步驟組嵌套循環(huán)調(diào)用組里套組、組里再套循環(huán)的完整示例見 tests/pipelines/call/nestedfor.yaml步驟組的調(diào)度核心在 pypyr/stepsrunner.pyrun_step_group負責取出組內(nèi)步驟逐個執(zhí)行成功/失敗回調(diào)on_success 與 on_failure流水線的收尾處理交給兩個內(nèi)置回調(diào)步驟組on_success所有步驟全部成功后執(zhí)行on_failure任意步驟拋錯時執(zhí)行然后原始錯誤繼續(xù)向上傳播steps: - name: pypyr.steps.echo in: echoMe: A - name: pypyr.steps.py in: py: raise ValueError(arb) # 這里故意失敗 - name: pypyr.steps.echo in: echoMe: 永遠不會執(zhí)行到 on_failure: - name: pypyr.steps.echo in: echoMe: 失敗回調(diào)記錄日志/發(fā)通知真實示例見 tests/pipelines/errors/fail-handler.yaml。回調(diào)機制的完整執(zhí)行邏輯在 pypyr/stepsrunner.py順序執(zhí)行所有指定步驟組全部成功 → 跑on_success中途出錯 → 跑on_failure失敗處理器自己出錯會被吞掉并記日志避免掩蓋真正的錯誤原因見 pypyr/stepsrunner.py失敗處理完畢原始異常繼續(xù)拋出——所以失敗回調(diào)適合記錄、通知、清理而不要指望它修復流水線進階技巧在on_failure里放一個pypyr.steps.stop步驟可以安靜終止流水線而不拋出紅色報錯常用于預期內(nèi)的失敗場景相關(guān)測試覆蓋在 tests/pipelines/errors/ 目錄下。命令行指定步驟組與回調(diào)一條命令跑法以上steps/on_success/on_failure都是默認值見 pypyr/pipeline.py命令行可以全部覆蓋參數(shù)定義見 pypyr/cli.py$ pypyr my-pipeline whoworld --groups steps sg1 --success done --failure cleanup參數(shù)默認含義--groupssteps要執(zhí)行的步驟組可指定多個--successon_success成功回調(diào)步驟組名--failureon_failure失敗回調(diào)步驟組名--log25日志級別10 打印完整堆棧這意味著你甚至可以在同一個流水線文件里定義多套流程用不同--groups組合調(diào)用而不必拆成多個文件。核心源碼文件導讀文件作用pypyr/dsl.py流水線 YAML 的領域模型步驟、循環(huán)、重試裝飾器的解析與執(zhí)行pypyr/stepsrunner.py步驟組調(diào)度器順序執(zhí)行 成功/失敗回調(diào)pypyr/pipeline.py單次流水線運行實例與默認組名配置pypyr/pipedef.py流水線定義體與加載器元數(shù)據(jù)pypyr/cli.py命令行入口與參數(shù)解析pypyr/steps/全部內(nèi)置步驟實現(xiàn)小結(jié)一張圖記住 pypyr 流水線 YAMLpipeline.yaml ├── context_parser 可選命令行參數(shù) → context 變量 ├── steps 默認步驟組 │ ├── name 步驟模塊必填 │ ├── in 步驟輸入?yún)?shù) │ ├── run / skip 條件執(zhí)行 │ ├── swallow 吞錯繼續(xù) │ ├── onError 自定義錯誤信息 │ ├── foreach / while / retry 循環(huán)與重試 │ └── description 步驟說明 ├── 自定義步驟組 可被 call 調(diào)用 / --groups 指定 ├── on_success 成功回調(diào) └── on_failure 失敗回調(diào)掌握了steps、步驟裝飾器、自定義步驟組和on_success/on_failure回調(diào)你就掌握了 pypyr 流水線 YAML 的全部核心語法——接下來只需瀏覽內(nèi)置步驟清單把真實的自動化流程串起來即可。【免費下載鏈接】pypyrpypyr task-runner cli api for automation pipelines. Automate anything by combining commands, different scripts in different languages applications into one pipeline process.項目地址: https://gitcode.com/gh_mirrors/py/pypyr創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考