Fast API 入門筆記 (二) - Typing Hint & Async

Posted by MingLun Allen Wu on Saturday, May 8, 2021

重溫舊夢: Fast API 入門筆記(一) - 基本介紹

Typing Hints

接續上一則筆記,我們繼續探討 Typing Hint 這個 FastAPI 特別著重的特點:

在實作上,Typing Hint 其實意味著:

在定義函式的參數時,連同型態一併定義

看看下面這個例子:

# 沒有實作 typing hint
@app.get("/users/{user_id}) 
def read_user(user_id, q): # 只放變數名稱
    return {"user_id": user_id, "q": q}

# 實作 typing hint
@app.get("/users/{user_id}) 
def read_user(user_id: int, q: Optional[str] = None): # 除了變數名稱,還設定變數的型態
    return {"user_id": user_id, "q": q}

實作 Typing Hint 後,在定義參數時需額外加上變數的型態 (user_id 後面會加上 int ),加上這些型態有幾項好處:

1. Error Checks, Completion

撰寫程式時檢查變數格式、自動補齊

這些功能並非 FastAPI 提供,在驗證變數型態時,主要是透過Pydantic來處理,所使用的變數型態都是 Python 原生的型態別,只需要在撰寫 Function 時加上變數型態即可。

當你遵照 Python 的Typing Hint 時,就能搭配 IDE 的工具來協助開發,避免語法上的錯誤或是設計時的遺漏。

舉例來說,撰寫 Typing Hint 後,就能透過 VScode 的 Extension 快速產生註解的 Template,很好用:

(根據定義的變數型態,自動產生註解的模板,大部分資訊都自動產生,開發者只需要填寫部分區塊即可,且模板的樣式可以自由選擇。)

在編寫程式時,儘管編譯器 「不知道」 這個函式的回傳內容是什麼,但是因為你已經事先定義這個回傳的「型態」,所以編譯器還是能夠提供當前這個型態可用的函式讓你使用.

 from datetime import datetime

def get_current_date() -> datetime: # 此處定義了此函式的回傳結果是datetime
    return datetime.now()

def get_current_date_error() -> datetime: # 此處定義了此函式的回傳結果是datetime
    return "2021-05-08" # 此處編譯器會跳出提醒訊息 (回傳結果應該要是datetime,但你回傳了str,是不是想搞事?)

current_date = get_current_date()

上面這段程式碼可以做到兩件事情:

  1. 開發時的型態檢查:

    在定義 get_current_date_error()時,我們順帶定義了回傳格式必須是 datetime,但是在function內部return時卻回傳了 str 格式,這時候編譯器能夠偵測到這裡有壞味道,並且提醒你。

  2. 函式智能感知(Intellisense):

    開發時自動提供可用的函式或是物件,以上方程式碼為例, current_date在沒有執行時,編譯器只知道它是個「變數」,沒辦法提供什麼建議。

    但是實作 Typing Hint後,編譯器知道這個變數是從 get_current_date()來的,而此函式的回傳結果是 datetime格式,所以當我們鍵入 current_date時,編譯器就能提供 datetime型態可用的函式供我們選用。

2.Data Conversion

自動將Request所夾帶的參數轉換(cast)為指定的格式

以下圖的API為例:

@app.get("/users/{user_id}")
def read_user(user_id: int):
    return {"user_id": user_id}

如果此時我們傳送了 127.0.0.1:8080/users/3 Request 至 Server,我們將會得到以下的回傳:

{"user_id":3}

發現了嗎? user_id 的 value 是 3 而不是 "3",原先在 Request 所夾帶的參數是字串格式,但是在設計函式時已經將 user_id 設定為整數,所以 FastAPI 會在處理過程中自動轉換格式。

3. Data Validation

檢查變數型態,若不符合,回傳「結構化」的錯誤訊息

如果此時發生了「無法轉換」的狀況呢?

例如傳送了: 127.0.0.1:8080/users/test Request, “test"是沒有辦法轉換成整數的,此時就會發生錯誤, FastAPI 將會回傳錯誤訊息。

{"detail":[
    {"loc":["path","user_id"],"msg":"value is not a valid integer","type":"type_error.integer"}
]}

回傳的格式是 Json 格式,意味著能夠判斷問題及預先處理,而不是如同 Flask500 Internal Server Error


FastAPI 實作

1. Import and Create Instance

使用 pip 安裝 FastAPI 後,就可以直接 Import,並且建立 FastAPI 的 Instance

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

2. 定義 API 路徑

FastAPIFlask 最大的差異在於定義 API 時,如何宣告使用的方法:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

FastAPI 在定義 API 時,透過 @app.[方法](路徑) 來決定,這點與 Flask 有點差異,但我自己覺得較直覺(也較符合 Restful API 的架構),常見的 HTTP Request 方法包含:

方法FastAPI 語法使用情境
POSTapp.post(路徑)建立資料
GETapp.get(路徑)取得資料
PUTapp.put(路徑)更新資料
DELETEapp.delete(路徑)刪除資料

3. 定義 Function

from fastapi import FastAPI

app = FastAPI()

@app.get("/hello_world")
async def hello_world():
    return {"message": "Hello World"}

定義 API 路徑後,接下來必須設計一個 Function,當 FastAPI 收到 GET 形式的 /hello_world 請求時,將會執行這個 Function.

這個Function 可以是 async function 或是 普通的 Function.

先暫停在這裏,何謂 Async Function?


Async Function

全名是 Asynchronize (非同步) Function,是在近期的 Python 中才出現的功能。

概念上:執行任務的過程中,有些任務不需要「計算」,例如讀取檔案、讀寫資料庫、呼叫別人的API,等待回應)時,在等待這些任務完成時,電腦的計算力是閒置的,可以先去做別的事情,等到其他任務完成,再回來繼續處理。

相對於 Asynchronize (非同步),較常見的稱為 Synchronize (同步)或稱為 Sequential (線性),我們舉個真實世界的例子來說明兩者的差異,假設今天我們要到速食店點餐:

  • Sequential: 我們看完菜單後向櫃檯人員點餐,點餐後後台開始製作餐點,這時候我們跟櫃檯人員大眼瞪小眼,也不做任何事情,靜靜地注視著彼此,直到餐點製作完成,櫃檯人員將餐點遞給你,開始享用美味的餐點。

同步處理在閒置時,只能持續等待到任務完成

  • Asynchronize: 我們看完菜單後向櫃檯人員點餐,點餐後後台開始製作餐點,我們拿到了一張號碼牌,我們開始滑手機,看看等等吃飽後要去哪裡吃甜點,當叫號聲響起時,發現是我們的餐點好了,這時候把手機收起來,前往領取餐點,開始享用美味的餐點。

非同步處理在閒置時,可以將注意力轉移到其他事情上

非同步處理相當適合 Web Application 的情境,在處理 Request 時,常需要跟 DB 或是其他 API 進行互動,等待的過程中會有許多「閒置」時間,在電腦的世界裡雖然都只需幾毫秒的等待,但是當同時有許多 Request 時,省下來的時間就相當可觀了。

這也是 FastAPIFlask 最大的差別,在 Flask 出現時,Python 還沒有支援異步處理的功能, Flask 要上線時,通常會再加上 WSGI (Web Service Gateway Interface) Server來增加服務的穩定性及速度,常見的 WSGI Server 是 gunicorn

FastAPI 採用的則是 ASGI Server (Asynchronize Service Gateway Interface) - uvicorn,在處理任務時透過 Asynchronize 的概念來提高執行的效率。

對於 WSGI Server 及 gunicron 有興趣的朋友,歡迎造訪我先前寫的另一篇文章:

Flask 想上線? 你還需要一些酷東西

在試圖了解 Asynchronize的過程中,我常常會將其與 Concurrency, Parallelism 搞混,相當推薦大家閱讀 FastAPI 官方對於 Async 的說明:

FastAPI - Concurrency and async / await

雖然使用非常多的 Emoji,閱讀起來有點混亂(xD),但用很貼近生活的實例來說明,相當值得一讀。


前往下集:

Fast API 入門筆記 (三) - Query Parameter & Path Parameter



See Also

監控資料品質的旅程