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