[Day 29] 定時任務與健康狀態檢查

發表於 更新於 分類於

最後想了想，決定來介紹一下怎麼寫「定時任務」和「健康狀態檢查 (health check)」這兩個小主題~

health check 翻成中文好不習慣…

定時任務

在某些時候，我們後端需要有一個定時執行的程式 (例如：向某個地方取得最新資料並更新到資料庫、清除過期 token)，此時我們就需要寫一個定時任務來幫我們完成這件事。

定時任務的實作方法也不只一種，例如：

1. 使用 APScheduler

APScheduler是一個可以寫排程的的套件，我們可以在啟動 FastAPI 的時候同時也初始化 APScheduler，後續再讓他自己定時完成任務，也可以透過 API 來取得或修改 APScheduler 的排程。

這樣的做法比較像是用 FastAPI 做介接，讓我們可以透過打 API 的方式去隔空操作。

2. 使用 Celery 的 celery beat

概念跟上面那個差不多，就是用 beat_schedule 設定定時任務，詳細可以參考這個文件。

3. 用 fastapi-utils 的 repeat_every

在開始介紹之前，先說明一下，這個好用模組已經有段時間沒更新了 (最後一次更新是 2023 年 3 月)，因此與最新版本的 FastAPI 相依的模組有落差，例如：fastapi-utils 尚未支援 pydantic 2.0、SQLAlchemy 2.0，儘管有多個 issue (issue 1、issue 2)希望可以支援新版，但作者似乎已經銷聲匿跡了…

但好在有其他熱心網友 fork 這個模組並更新支援的模組，例如：FastApi-RESTful、fastapi-utilities，因此大家在使用時可以考慮一下到底要安裝哪個套件。

這是我個人最喜歡的寫法，因為這個最簡潔，畢竟我只要定時啟動就好，不需要其他功能。

fastapi-utils 是一個專門提供 FastAPI 相關功能的模組 (相信從名稱就可以感受得出來)，它裡面就有 repeat_every 這個 decorator 可以幫我們做到定時任務的設定。

首先，先安裝 fastapi-utils

1
pip install fastapi-utils

或是 FastApi-RESTful

1
pip install fastapi-restful[all]

接著看一下這個範例 (這邊用 FastApi-RESTful 當例子)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# main.py
from fastapi import FastAPI
# from fastapi_utils.tasks import repeat_every
from fastapi_restful.tasks import repeat_every

app = FastAPI()

@repeat_every(seconds=5)
async def repeat_task() -> None:
    print("test")

@app.on_event("startup")
async def startup_event():
    await repeat_task()

使用時，只要在執行的定時任務用 repeat_every 修飾即可，如果需要其他設定則可以參考文件。此外，通常我們也會希望啟動服務的時候也要執行一次，因此可以放進 startup 事件中 (或是用較新的 lifespan 寫法也可以)。

相信大家應該有感受到這個做法的簡潔吧 XD

健康狀態檢查

有時候，我們需要確認後端是否正常運作，這件事的時機可能是在部屬後 (快速確認是否部屬成功)，也有可能是定時觀察 (避免服務掛了都不知道 XD)。最簡單的做法就是打一個最簡單的 API (或是故意寫一個 API 專門回傳極簡單的訊息)，看有沒有如預期地拿到正確回應就好。

但是，如果我們的服務更複雜一些，還有接資料庫，那我們就需要再多確認後端是否可以連上資料庫。這件事也不難，就在 API 內加上一個 Depends()，讓 API 取得與資料庫連線的 Session (詳細可以參考 [Day 15])。

但是 (第二次 XD)，如果服務再更複雜，有更多東西需要檢查，那這個 health check 用的 API 就會越來越複雜，這時，可以考慮使用 fastapi-health 這個套件。

這邊直接看官方範例

1
2
3
4
5
6
7
8
9
10
11
from fastapi import FastAPI, Depends
from fastapi_health import health

def get_session():
    return True

def is_database_online(session: bool = Depends(get_session)):
    return session

app = FastAPI()
app.add_api_route("/health", health([is_database_online]))

它的核心用法就是用 add_api_route() 的第二個變數這邊使用 health 這個函數，我們只要把要檢查的項目放進 list 傳給 health 函數就好了。其他進階設定就請大家再去看文件說明了 (一點點而已)。

補充

最後還是想補充一下，如果是檢查網站是否正常，或是後端的 API 有暴露在外的話，可以考慮使用 UptimeRobot 這個免費工具 (在 [Day 26] 有稍微提到過)，他除了可以定時進行 health check 之外，也可以再發現問題的時候發送通知，通知的方式有很多種，從基本的 email 到常見的通訊軟體都可以 (透過 webhook)，使用起來很方便~

小結

今天介紹了兩個小主題，在「定時任務」我們介紹了

  1. 多個實作的方法
  2. fastapi-utils 的範例

在「health check」則是簡單的介紹了 fastapi-health 這個套件並補充介紹 UptimeRobot 這個工具

終於只剩下最後一篇了，雖然還有很多主題可以寫，但明天還是用心得來收尾好了 XD