如何用 Plombery 创建你的第一条 Pipeline？Task、Pipeline、Trigger 三大核心概念一次讲透

如何用 Plombery 创建你的第一条 PipelineTask、Pipeline、Trigger 三大核心概念一次讲透【免费下载链接】plomberyPython task scheduler with a user-friendly web UI项目地址: https://gitcode.com/gh_mirrors/pl/plomberyPlombery 是一款带友好 Web UI 的 Python 任务调度器task scheduler你用几行 Python 代码定义 Pipeline它自动帮你调度、记录日志并提供可视化界面查看每次运行Run的状态与耗时。本文将带你从零创建你的第一条 Pipeline并彻底讲透 Task、Pipeline、Trigger 三大核心概念。快速上手10 分钟安装 Plombery只需 Python 3.8 即可运行 Plombery。官方推荐的完整安装步骤见 docs/get-started.md创建并激活虚拟环境python -m venv .venv安装 Plomberypip install plombery安装 ASGI 服务器Plombery 基于 FastAPIpip install uvicorn安装完成后你离第一条 Pipeline 只差一个 Python 文件。核心概念 1Task —— 一个带装饰器的 Python 函数Task 是 Plombery 的最小积木任何普通 Python 函数加上task装饰器就成为一个任务。它支持同步和异步写法函数定义见 src/plombery/pipeline/task.pyfrom plombery import task, get_logger task async def fetch_raw_sales_data(): logger get_logger() # 日志会自动收集并展示在 Web UI logger.info(Fetching sales data...) return {price: 99, store_id: 3} # 返回值会传给下一个 Task两个关键细节日志get_logger()必须在 task 函数内部调用日志会自动汇总到 Web UI 的运行日志页面数据传递Pipeline 内的 Task 按顺序执行前一个 Task 的返回值会作为位置参数传给后续 Task天然形成数据流核心概念 2Pipeline —— Task 的有序编排Pipeline 是 1 个或多个 Task 组成的执行序列通过register_pipeline注册。最小完整示例只需几行完整教程见 docs/create-a-pipeline.mdfrom plombery import task, register_pipeline task async def fetch_data(): return get_my_things() register_pipeline( idmy_pipeline, # 必填唯一标识 tasks[fetch_data], # 必填Task 列表 )除了id和tasks你还可以声明paramsPydantic 输入参数Web UI 会自动生成参数表单、name、description等详见 src/plombery/pipeline/pipeline.py。核心概念 3Trigger —— Pipeline 的启动入口Trigger 是运行 Pipeline 的入口可以是一个定时计划schedule也可以是 Web UI 上的按钮或 HTTP 接口实现见 src/plombery/pipeline/trigger.py。即使不声明任何 Trigger系统也会自动附加一个Pipeline trigger给你两个免费入口手动按钮首页和 Pipeline 页面都有 Run 按钮一键触发HTTP 接口Pipeline 页面展示 Run URLPOST 请求即可程序化运行定时调度是 Plombery 最常用能力。注册 Pipeline 时传入 Trigger 列表schedule参数支持 APScheduler 的 Cron、Interval、Date 等触发器例如每天运行一次from apscheduler.triggers.interval import IntervalTrigger from plombery import Trigger triggers[ Trigger( iddaily, nameDaily, descriptionRun the pipeline every day, scheduleIntervalTrigger(days1), ), ],每个 Trigger 还能绑定专属params——相当于为 Pipeline 定义多个固定参数的快捷入口适合让不熟悉代码的同事也能安全地手动运行。完整参数用法可参考项目内置示例 examples/src/sales_pipeline.py。运行你的第一条 Pipeline 并查看结果在入口文件底部启动服务if __name__ __main__: import uvicorn uvicorn.run(plombery:get_app, reloadTrue, factoryTrue)执行python src/app.py浏览器打开http://localhost:8000即可看到Trigger 详情页调度计划、下次运行时间、触发器健康度近期成功/失败、平均耗时趋势图Runs 列表每一次 Pipeline Run 的状态、开始时间、耗时运行日志按 Task、日志级别筛选的详细日志项目examples/目录还提供了更贴近实战的示例比如监控 SSL 证书到期时间的 Pipeline效果如下相关源码位于 examples/src/ssl_certificates.py。总结一句话记住三大概念概念一句话定义Task加了task装饰器的 Python 函数是 Pipeline 的基本积木Pipeline若干 Task 组成的执行序列用register_pipeline注册Trigger运行 Pipeline 的入口定时计划、HTTP 接口或 Web UI 按钮现在你已掌握用 Plombery 创建 Pipeline 的完整思路定义 Task → 编排 Pipeline → 配置 Trigger → 在 Web UI 上观察运行。接下来不妨去 docs/triggers.md 和 docs/tasks.md 探索更多玩法把日常的数据同步、报表生成任务都交给它吧【免费下载链接】plomberyPython task scheduler with a user-friendly web UI项目地址: https://gitcode.com/gh_mirrors/pl/plombery创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考