pypyr流水线YAML语法完全解析步骤、步骤组与成功/失败回调一次看懂【免费下载链接】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顺序执行。本文完整解析 pypyr 流水线的 YAML 语法每个步骤step支持哪些字段、如何用步骤组step group拆分可复用流程以及on_success/on_failure两大回调如何兜底——一篇看懂直接上手。为什么用 YAML 写自动化pypyr 的定位传统 shell 脚本或 Makefile 写多了都知道头疼换平台就崩、变量传递混乱、错误处理靠硬扛。pypyr 的思路很直接——用一个 YAML 文件描述整条自动化流程然后一条命令跑完$ pip install --upgrade pypyr $ pypyr my-pipeline # 文件名不带 .yaml✅ 跨平台Linux / macOS / Windows 通吃CI 服务器和容器里同样能跑✅ 多语言混编同一条流水线里可以跑 shell 命令、Python 代码、调外部程序✅ 流程可控循环、条件、重试、失败兜底全部在 YAML 里声明想深入了解项目结构可从 pypyr/pipeline.py 开始它定义了一次流水线运行的运行时对象。流水线文件的最小骨架一个 pypyr 流水线 YAML 的最小形态只需要一个steps键# my-pipeline.yaml steps: - name: pypyr.steps.echo in: echoMe: 你好pypyr 流水线执行pypyr my-pipeline即可看到输出。如果想让流水线接收命令行参数加上可选的context_parsercontext_parser: pypyr.parser.keyvaluepairs steps: - name: pypyr.steps.echo in: echoMe: hello {who}随后pypyr my-pipeline whoworldYAML 里的{who}就会被自动替换成world——这就是 pypyr 的变量替换变量来自 context 上下文。步骤的两种写法简单与复杂简单步骤一行搞定步骤名就是步骤模块的完整名称没有参数时直接写一行steps: - pypyr.steps.echo - pypyr.steps.now复杂步骤name in 参数需要传参时用字典形式核心就两个字段steps: - name: pypyr.steps.cmd # 要执行的步骤模块 in: # 注入步骤的输入参数 cmd: echo 任意命令in参数的生命周期在源码里写得很清楚执行前注入 context执行完自动移除不会污染后续步骤见 pypyr/dsl.py。内置步骤如pypyr.steps.echo、pypyr.steps.cmd、pypyr.steps.py都位于 pypyr/steps/ 目录每个文件名就是一个步骤。步骤装饰器速查表一次看懂 7 个字段复杂步骤除了name和in还支持这些装饰器字段解析逻辑见 pypyr/dsl.py 的_init_from_dict字段默认值作用runTrue表达式为真才执行支持变量替换skipFalse表达式为真则跳过本步骤swallowFalse出错时吞掉异常流水线继续走onError无出错时附加的自定义错误信息description无运行时在控制台打印的步骤说明foreach无对列表逐项循环执行retry无失败自动重试可配退避策略while无条件成立则反复循环执行条件执行是最常用的场景例如steps: - name: pypyr.steps.cmd run: {deployEnabled} # 只有 deployEnabled 为 True 才执行 in: cmd: echo 执行部署容错场景用swallow某步骤失败不想中断整条流水线时- name: pypyr.steps.cmd swallow: True in: cmd: 可能失败的命令循环与重试让流水线自己跑foreach 循环列表里每项执行一遍当前项通过{i}引用steps: - name: pypyr.steps.cmd foreach: [once, twice, thrice] in: cmd: echo say {i}while 循环条件满足就重复跑内置{whileCounter}记录轮次max限制最多轮数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] # 调用同文件里的步骤组 sg1 sg1: # 自定义步骤组 - name: pypyr.steps.echo in: echoMe: 我是被调用的 sg1steps组默认最先执行pypyr.steps.call可跳转调用其他命名步骤组嵌套循环调用组里套组、组里再套循环的完整示例见 tests/pipelines/call/nestedfor.yaml步骤组的调度核心在 pypyr/stepsrunner.pyrun_step_group负责取出组内步骤逐个执行成功/失败回调on_success 与 on_failure流水线的收尾处理交给两个内置回调步骤组on_success所有步骤全部成功后执行on_failure任意步骤抛错时执行然后原始错误继续向上传播steps: - name: pypyr.steps.echo in: echoMe: A - name: pypyr.steps.py in: py: raise ValueError(arb) # 这里故意失败 - name: pypyr.steps.echo in: echoMe: 永远不会执行到 on_failure: - name: pypyr.steps.echo in: echoMe: 失败回调记录日志/发通知真实示例见 tests/pipelines/errors/fail-handler.yaml。回调机制的完整执行逻辑在 pypyr/stepsrunner.py顺序执行所有指定步骤组全部成功 → 跑on_success中途出错 → 跑on_failure失败处理器自己出错会被吞掉并记日志避免掩盖真正的错误原因见 pypyr/stepsrunner.py失败处理完毕原始异常继续抛出——所以失败回调适合记录、通知、清理而不要指望它修复流水线进阶技巧在on_failure里放一个pypyr.steps.stop步骤可以安静终止流水线而不抛出红色报错常用于预期内的失败场景相关测试覆盖在 tests/pipelines/errors/ 目录下。命令行指定步骤组与回调一条命令跑法以上steps/on_success/on_failure都是默认值见 pypyr/pipeline.py命令行可以全部覆盖参数定义见 pypyr/cli.py$ pypyr my-pipeline whoworld --groups steps sg1 --success done --failure cleanup参数默认含义--groupssteps要执行的步骤组可指定多个--successon_success成功回调步骤组名--failureon_failure失败回调步骤组名--log25日志级别10 打印完整堆栈这意味着你甚至可以在同一个流水线文件里定义多套流程用不同--groups组合调用而不必拆成多个文件。核心源码文件导读文件作用pypyr/dsl.py流水线 YAML 的领域模型步骤、循环、重试装饰器的解析与执行pypyr/stepsrunner.py步骤组调度器顺序执行 成功/失败回调pypyr/pipeline.py单次流水线运行实例与默认组名配置pypyr/pipedef.py流水线定义体与加载器元数据pypyr/cli.py命令行入口与参数解析pypyr/steps/全部内置步骤实现小结一张图记住 pypyr 流水线 YAMLpipeline.yaml ├── context_parser 可选命令行参数 → context 变量 ├── steps 默认步骤组 │ ├── name 步骤模块必填 │ ├── in 步骤输入参数 │ ├── run / skip 条件执行 │ ├── swallow 吞错继续 │ ├── onError 自定义错误信息 │ ├── foreach / while / retry 循环与重试 │ └── description 步骤说明 ├── 自定义步骤组 可被 call 调用 / --groups 指定 ├── on_success 成功回调 └── on_failure 失败回调掌握了steps、步骤装饰器、自定义步骤组和on_success/on_failure回调你就掌握了 pypyr 流水线 YAML 的全部核心语法——接下来只需浏览内置步骤清单把真实的自动化流程串起来即可。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考