Rocketry

我原本的排程系統都是使用APScheduler，偶然間看到Rocketry，稍微看了一下官方文件後，就開始著手進行學習，套件使用方法相當類似於FastAPI，開發者也說是被FastAPI所啟發。這位開發者其實也是在金融相關的公司工作，所以他所設計的就是那麼對味。另一個金融公司開發的就是pandas。

Why Rocketry? — Rocketry

Rocketry特色

文字描述模式

這個功能是比較少見的模式，利用文字敘述想要執行任務時間或頻率，開放的模式也代表著相對不嚴謹，所以我是比較少用一些。

@app.task('every 10 seconds')
def do_constantly():
    ...

@app.task('every 1 minute')
def do_minutely():
    ...

@app.task('every 1 hour')
def do_hourly():
    ...

@app.task('every 1 day')
def do_daily():
    ...

@app.task('every 2 days 2 hours 20 seconds')
def do_custom():
    ...

提供Pipeline架構

一般的任務排程系統主要由時間驅動，並可以使用預設的固定時點(例如每日上午10點)，或是時間區間執行(例如每五分鐘），但Rocketry提供了一個pipeline的模式，當任務流程成功與失敗時都可以由客製化的函數來執行。

例如下方官方文件中的範例程式，可以很直覺得看到當某一任務完成後，要執行的任務，或失敗執行的任務，甚至還可以多個都完成，或其中一個失敗的定義。例如我們就可以定義的任一函數失敗時，寄送Email進行通知。

from rocketry.conds import after_success, after_fail, after_finish

@app.task()
def do_things():
    ...

@app.task(after_success(do_things))
def do_after_success():
    ...

@app.task(after_fail(do_things))
def do_after_fail():
    ...

@app.task(after_finish(do_things))
def do_after_fail_or_success():
    ...

而Pipeline之間就一定會有資料的傳輸，例如取得某個資料後，將原始資料進行某種轉換或再計算，並整理出適合存入資料庫的格式，再進行資料庫更新。

from rocketry.conds import daily, after_success
from rocketry.args import Return

@app.task(daily)
def do_first():
    return 'Hello World'

@app.task(after_success(do_first))
def do_second(arg=Return(do_first)):
    # arg's value is "Hello World"
    ...

這個Pipeline概念是我最喜歡使用Rocketry的其中一個原因！

---

Rocketry安裝問題

如果直接使用pip install rocketry，而運行基本程序後，將會出現一堆Pydantic的錯誤，這時候只能將Pydantic重新以1.10.10版本進行安裝，這個問題才能解決，可能要等開發團隊更新Pydantic 2.X.X版以後才能運行。

Rocketry + Full-Stack (FastAPI)

AirFlow的網頁介面實在是相當方便，Rocketry開發者也有提供一個與FastAPI整合的專案，可以透過網頁介面管理與操作任務。

GitHub - Miksus/rocketry-with-fastapi: Example of how to create a scheduler with an APIGitHub