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

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

【免费下载链接】plomberyPython task scheduler with a user-friendly web UI项目地址: https://gitcode.com/gh_mirrors/pl/plombery

Plombery 是一款带友好 Web UI 的 Python 任务调度器（task scheduler）：你用几行 Python 代码定义 Pipeline，它自动帮你调度、记录日志，并提供可视化界面查看每次运行（Run）的状态与耗时。本文将带你从零创建你的第一条 Pipeline，并彻底讲透 Task、Pipeline、Trigger 三大核心概念。

快速上手：10 分钟安装 Plombery

只需 Python 3.8+ 即可运行 Plombery。官方推荐的完整安装步骤见 docs/get-started.md：

1. 创建并激活虚拟环境：python -m venv .venv
2. 安装 Plombery：pip install plombery
3. 安装 ASGI 服务器（Plombery 基于 FastAPI）：pip install uvicorn

安装完成后，你离第一条 Pipeline 只差一个 Python 文件。

核心概念 1：Task —— 一个带装饰器的 Python 函数

Task 是 Plombery 的最小积木：任何普通 Python 函数，加上@task装饰器，就成为一个任务。它支持同步和异步写法（函数定义见 src/plombery/pipeline/task.py）：

from 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，天然形成数据流

核心概念 2：Pipeline —— Task 的有序编排

Pipeline 是 1 个或多个 Task 组成的执行序列，通过register_pipeline注册。最小完整示例只需几行（完整教程见 docs/create-a-pipeline.md）：

from plombery import task, register_pipeline @task async def fetch_data(): return get_my_things() register_pipeline( id="my_pipeline", # 必填：唯一标识 tasks=[fetch_data], # 必填：Task 列表 )

除了id和tasks，你还可以声明params（Pydantic 输入参数，Web UI 会自动生成参数表单）、name、description等，详见 src/plombery/pipeline/pipeline.py。

核心概念 3：Trigger —— Pipeline 的启动入口

Trigger 是运行 Pipeline 的入口：可以是一个定时计划（schedule），也可以是 Web UI 上的按钮或 HTTP 接口（实现见 src/plombery/pipeline/trigger.py）。

📌即使不声明任何 Trigger，系统也会自动附加一个"Pipeline trigger"，给你两个免费入口：

• 手动按钮：首页和 Pipeline 页面都有 Run 按钮，一键触发
• HTTP 接口：Pipeline 页面展示 Run URL，POST 请求即可程序化运行

🕘定时调度是 Plombery 最常用能力。注册 Pipeline 时传入 Trigger 列表，schedule参数支持 APScheduler 的 Cron、Interval、Date 等触发器，例如"每天运行一次"：

from apscheduler.triggers.interval import IntervalTrigger from plombery import Trigger triggers=[ Trigger( id="daily", name="Daily", description="Run the pipeline every day", schedule=IntervalTrigger(days=1), ), ],

每个 Trigger 还能绑定专属params——相当于为 Pipeline 定义多个"固定参数的快捷入口"，适合让不熟悉代码的同事也能安全地手动运行。完整参数用法可参考项目内置示例 examples/src/sales_pipeline.py。

运行你的第一条 Pipeline 并查看结果

在入口文件底部启动服务：

if __name__ == "__main__": import uvicorn uvicorn.run("plombery:get_app", reload=True, factory=True)

执行python src/app.py，浏览器打开http://localhost:8000即可看到：

• Trigger 详情页：调度计划、下次运行时间、触发器健康度（近期成功/失败）、平均耗时趋势图
• Runs 列表：每一次 Pipeline Run 的状态、开始时间、耗时
• 运行日志：按 Task、日志级别筛选的详细日志

项目examples/目录还提供了更贴近实战的示例，比如监控 SSL 证书到期时间的 Pipeline，效果如下：

相关源码位于 examples/src/ssl_certificates.py。

总结：一句话记住三大概念

现在你已掌握用 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