最後想了想,決定來介紹一下怎麼寫「定時任務」和「健康狀態檢查 (health check)」這兩個小主題~
health check 翻成中文好不習慣...
在某些時候,我們後端需要有一個定時執行的程式 (例如:向某個地方取得最新資料並更新到資料庫、清除過期 token),此時我們就需要寫一個定時任務來幫我們完成這件事。
定時任務的實作方法也不只一種,例如:
APScheduler是一個可以寫排程的的套件,我們可以在啟動 FastAPI 的時候同時也初始化 APScheduler,後續再讓他自己定時完成任務,也可以透過 API 來取得或修改 APScheduler 的排程。
這樣的做法比較像是用 FastAPI 做介接,讓我們可以透過打 API 的方式去隔空操作。
概念跟上面那個差不多,就是用 beat_schedule
設定定時任務,詳細可以參考這個文件。
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
pip install fastapi-utils
或是 FastApi-RESTful
pip install fastapi-restful[all]
接著看一下這個範例 (這邊用 FastApi-RESTful
當例子)
# 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
這個套件。
這邊直接看官方範例
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),使用起來很方便~
今天介紹了兩個小主題,在「定時任務」我們介紹了
fastapi-utils
的範例在「health check」則是簡單的介紹了 fastapi-health
這個套件並補充介紹 UptimeRobot 這個工具
終於只剩下最後一篇了,雖然還有很多主題可以寫,但明天還是用心得來收尾好了 XD