Prefect + pushplus：Flow 失败推到微信

效果：Prefect flow 进入失败状态时，on_failure 钩子用 requests 或 curl 调 pushplus。

前置条件

  • 一个 pushplus token（官网扫码获取）
  • 可改 flow 代码，运行环境已装 requests（或能调 curl）

配置步骤

import os
import requests
from prefect import flow

def notify_failure(flow_obj, flow_run, state):
    requests.post(
        "https://www.pushplus.plus/send",
        json={
            "token": os.environ["PUSHPLUS_TOKEN"],
            "title": f"Prefect 失败: {flow_obj.name}",
            "content": f"run={flow_run.id}\nstate={state}\nmsg={state.message}",
            "template": "markdown",
        },
        timeout=10,
    )

@flow(on_failure=[notify_failure])
def my_flow():
    raise RuntimeError("demo")

token 用运行环境的 PUSHPLUS_TOKEN，不要写进仓库。字段名以你使用的 Prefect 2/3 API 为准。

不方便改 Python 时，在部署包装脚本里：

python run_flow.py || curl -s -X POST "https://www.pushplus.plus/send" \
  -H "Content-Type: application/json" \
  -d "{\"token\":\"$PUSHPLUS_TOKEN\",\"title\":\"Prefect 失败\",\"content\":\"请查看 worker 日志\",\"template\":\"markdown\"}"

验证

curl -X POST "https://www.pushplus.plus/send" \
  -H "Content-Type: application/json" \
  -d '{"token":"你的token","title":"Prefect 配置测试","content":"token 可用","template":"markdown"}'

再跑一条必失败的 flow，确认钩子触发且接口 code 为 200。

常见问题

钩子没跑？ 确认装饰器是 on_failure 且异常未在 task 内被吞掉；worker 日志里应有 requests 调用。

成功也通知？ 不要用 on_completion 除非你确实需要每次结束都推。

更新时间: