FastAPI：Python 世界里最受歡迎的異步框架

楔子

這次我們來聊一聊 FastAPI，它和我們之前介紹的 Sanic 是類似的，都是 Python 中的異步 web 框架。相比 Sanic，FastAPI 更加的成熟、社區也更加的活躍，那么 FastAPI 都有哪些特點呢？

• 快速：擁有非常高的性能，歸功於 Starlette 和 Pydantic；Starlette 用於路由匹配，Pydantic 用於數據驗證
• 開發效率：功能開發效率提升 200% 到 300%
• 減少 bug：減少 40% 的因為開發者粗心導致的錯誤
• 智能：內部的類型注解非常完善，編輯器可處處自動補全
• 簡單：框架易於使用，文檔易於閱讀
• 簡短：使代碼重復最小化，通過不同的參數聲明實現豐富的功能
• 健壯：可以編寫出線上使用的代碼，並且會自動生成交互式文檔
• 標准化：兼容 API 相關開放標准

FastAPI 最大的特點就是它使用了 Python 的類型注解，我們后面會詳細說，下面來安裝一下 FastAPI。

使用 FastAPI 需要 Python 版本大於等於 3.6。

首先是 pip install fastapi，會自動安裝 Starlette 和 Pydantic；然后還要 pip install uvicorn，因為 uvicorn 是運行相關應用程序的服務器。或者一步到胃：pip install fastapi[all]，會將所有依賴全部安裝。

請求與響應

我們來使用 FastAPI 編寫一個簡單的應用程序：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

# 類似於 app = Flask(__name__)
app = FastAPI()

# 綁定路由和視圖函數
@app.get("/")
async def index():
    return {"name": "古明地覺"}


# 在 Windows 中必須加上 if __name__ == "__main__"，否則會拋出 RuntimeError: This event loop is already running
if __name__ == "__main__":
    # 啟動服務，因為我們這個文件叫做 main.py，所以需要啟動 main.py 里面的 app
    # 第一個參數 "main:app" 就表示這個含義，然后是 host 和 port 表示監聽的 ip 和端口
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

整個過程顯然很簡單，然后我們我們在瀏覽器中輸入 "localhost:5555" 就會顯示相應的輸出，我們看到在視圖函數中可以直接返回一個字典。當然除了字典，其它的數據類型也是可以的，舉個栗子：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/int")
async def index1():
    return 666

@app.get("/str")
async def index2():
    return "古明地覺"

@app.get("/bytes")
async def index3():
    return b"satori"

@app.get("/tuple")
async def index4():
    return ("古明地覺", "古明地戀", "霧雨魔理沙")

@app.get("/list")
async def index5():
    return [{"name": "古明地覺", "age": 17}, {"name": "古明地戀", "age": 16}]


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

這里我們直接使用 requests 發請求：

我們看到基本上都是支持的，只不過元組自動轉成列表返回了。這里我們在路由中指定了路徑，可以看到 FastAPI 中的路徑形式和其它框架並無二致，只不過目前的路徑是寫死的，如果我們想動態聲明路徑參數該怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/items/{item_id}")
async def get_item(item_id):
    """和 Flask 不同，Flask 是使用 <>，而 FastAPI 使用 {}"""
    return {"item_id": item_id}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

整體非常簡單，路由里面的路徑參數我們可以放任意個，只是 {} 里面的參數必須要在定義的視圖函數的參數中出現。但是問題來了，我們好像沒有規定類型啊，如果我們希望某個路徑參數只能接收指定的類型要怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/apple/{item_id}")
async def get_item(item_id: int):
    """和 Flask 不同，Flask 定義類型是在路由當中，也就是在 <> 里面，變量和類型通過 : 分隔
       而 FastAPI 是使用類型注解的方式，此時的 item_id 要求一個整型（准確的說是一個能夠轉成整型的字符串）"""
    return {"item_id": item_id}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

如果我們傳遞的值無法轉成整型的話，那么會進行提示：告訴我們 value 不是一個有效的整型，可以看到給的提示信息還是非常清晰的。

所以通過 Python 的類型聲明，FastAPI提供了數據校驗的功能，當校驗不通過的時候會清楚地指出沒有通過的原因。在我們開發和調試的時候，這個功能非常有用。

交互式文檔

FastAPI 會自動提供一個類似於 Swagger 的交互式文檔，我們輸入 "localhost:5555/docs" 即可進入。

有興趣可以自己嘗試測試一下，然后我們注意一下里面的 /openapi.json，我們可以點擊進去，會發現里面包含了我們定義的路由信息。

至於 "localhost:5555/docs" 頁面本身，我們也是可以進行設置的：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI(title="測試文檔",
              description="這是一個簡單的 demo",
              docs_url="/my_docs",
              openapi_url="/my_openapi")

@app.get("/apple/{item_id}")
async def get_item(item_id: int):
    """和 Flask 不同，Flask 定義類型是在路由當中，也就是在 <> 里面，變量和類型通過 : 分隔
       而 FastAPI 是使用類型注解的方式，此時的 item_id 要求一個整型（准確的說是一個能夠轉成整型的字符串）"""
    return {"item_id": item_id}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

然后我們再重新進入，此時就需要通過 "localhost:5555/my_docs"：

整體沒什么難度，我們還可以指定其它參數，比如 version 表示版本，可以自己試試。該頁面主要用來測試自己編寫的 API 服務，不過個人更喜歡使用 requests 發請求。

路由順序

然后我們在定義路由的時候需要注意一下順序，舉個栗子：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/users/me")
async def read_user_me():
    return {"user_id": "the current user"}

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

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

因為路徑操作是按照順序進行的，所以這里要保證 /users/me 在 /users/{user_id} 的前面，否則的話只會匹配到 /users/{user_id}，此時如果訪問 /users/me，那么會返回一個解析錯誤，因為字符串 "me" 無法解析成整型。

使用枚舉

我們可以將某個路徑參數通過類型注解的方式聲明為指定的類型（准確的說是可以轉成指定的類型，因為默認都是字符串），但如果我們希望它只能是我們規定的幾個值之一該怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from enum import Enum
from fastapi import FastAPI
import uvicorn

app = FastAPI()

class Name(str, Enum):
    satori = "古明地覺"
    koishi = "古明地戀"
    marisa = "霧雨魔理沙"

@app.get("/users/{user_name}")
async def get_user(user_name: Name):
    return {"user_id": user_name}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

通過枚舉的方式可以實現這一點，我們來測試一下：

結果和我們期望的是一樣的，我們可以再來看看 docs 生成的文檔：

提示我們，可以用的值：古明地覺、古明地戀、霧雨魔理沙。

路徑中包含 /

假設我們有這樣一個路由：/files/{file_path}，而用戶傳遞的 file_path 中顯然是可以帶 / 的，假設 file_path 是 /root/test.py，那么路由就變成了 /files//root/test.py，顯然這是有問題的。

那么為了防止解析出錯，我們需要做一個類似於 Flask 中的操作：

from fastapi import FastAPI
import uvicorn

app = FastAPI()


# 聲明 file_path 的類型為 path，這樣它會被當成一個整體
@app.get("/files/{file_path:path}")
async def get_file(file_path: str):
    return {"file_path": file_path}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

然后我們來訪問一下：

查詢參數

查詢參數在 FastAPI 中依舊可以通過類型注解的方式進行聲明，如果函數中定義了不屬於路徑參數的參數時，那么它們將會被自動解釋會查詢參數。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/user/{user_id}")
async def get_user(user_id: str, name: str, age: int):
    """我們在函數中參數定義了 user_id、name、age 三個參數
       顯然 user_id 和 路徑參數中的 user_id 對應，然后 name 和 age 會被解釋成查詢參數
       這三個參數的順序沒有要求，但是一般都是路徑參數在前，查詢參數在后
    """
    return {"user_id": user_id, "name": name, "age": age}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

注意：name 和 age 是沒有默認值的，這意味着它們是必須要傳遞的，否則報錯。

我們看到當不傳遞 name 和 age 的時候，會直接提示你相關的錯誤信息。如果我們希望用戶可以不傳遞的話，那么必須要指定一個默認值。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/user/{user_id}")
async def get_user(user_id: str, name: str = "UNKNOWN", age: int = 0):
    return {"user_id": user_id, "name": name, "age": age}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

對於查詢參數，由於它們指定了類型，所以我們也要傳遞正確類型的數據。假設這里的 age 傳遞了一個 "abc"，那么也是通不過的。

如果默認值和類型不相同怎么辦？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/user/{user_id}")
async def get_user(user_id: str, name: str = "UNKNOWN", age: int = "哈哈哈"):
    return {"user_id": user_id, "name": name, "age": age}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

這里的 age 需要接收一個整型，但是默認值卻是一個字符串，那么此時會有什么情況發生呢？我們來試一下：

我們看到，傳遞的 age 依舊需要整型，只不過在不傳的時候會使用字符串類型的默認值，所以類型和默認值類型不同的時候也是可以的，只不過這么做顯然是不合理的。但是問題來了，我們可不可以指定多個類型呢？比如 user_id 按照整型解析、解析不成功退化為字符串。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Union, Optional
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/user/{user_id}")
async def get_user(user_id: Union[int, str], name: Optional[str] = None):
    """通過 Union 來聲明一個混合類型，int 在前、str 在后。會先按照 int 解析，解析失敗再變成 str
       然后是 name，它表示字符串類型、但默認值為 None（不是字符串），那么應該聲明為 Optional[str]"""
    return {"user_id": user_id, "name": name}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

所以 FastAPI 的設計還是非常不錯的，通過 Python 的類型注解來實現參數類型的限定可以說是非常巧妙的，因此這也需要我們熟練掌握 Python 的類型注解。

bool 類型自動轉換

對於布爾類型，FastAPI 支持自動轉換，舉個栗子：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/{flag}")
async def get_flag(flag: bool):
    return {"flag": flag}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

多個路徑和查詢參數

我們之前說過，可以定義任意個路徑參數，只要動態的路徑參數{} 里面的 在函數的參數中都出現即可。當然查詢參數也可以是任意個，FastAPI 可以處理的很好。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/postgres/{schema}/v1/{table}")
async def get_data(schema: str,
                   table: str,
                   select: str = "*",
                   where: Optional[str] = None,
                   limit: Optional[int] = None,
                   offset: Optional[int] = None):
    """標准格式是：路徑參數按照順序在前，查詢參數在后
       但其實對順序是沒有什么要求的"""
    query = f"select {select} from {schema}.{table}"
    if where:
        query += f" where {where}"
    if limit:
        query += f" limit {limit}"
    if offset:
        query += f" offset {offset}"
    return {"query": query}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

然后我們使用 requests 來測試一下：

print(
    requests.get("http://localhost:5555/postgres/ods/v1/staff").json()
)  # {'query': 'select * from ods.staff'}
print(
    requests.get("http://localhost:5555/postgres/ods/v1/staff?select=id, name&where=id > 3&limit=100").json()
)  # {'query': 'select id, name from ods.staff where id > 3 limit 100'}

Depends

這個老鐵比較特殊，它是用來做什么的呢？我們來看一下：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional
import uvicorn
from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    # common_parameters 接收三個參數：q、skip、limit
    # 然后在解析請求的時候，會將 q、skip、limit 傳遞到 common_parameters 中，然后將返回值賦值給 commons
    # 但如果解析不到某個參數時，那么會判斷函數中參數是否有默認值，沒有的話就會返回錯誤，而不是傳遞一個 None 進去
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons


if __name__ == "__main__":
    uvicorn.run("main1:app", host="0.0.0.0", port=5555)

我們來測試一下：

所以 Depends 能夠很好的實現依賴注入，而且我們特意寫了兩個路由，就是想表示它們是彼此獨立的。因此當有共享的邏輯、或者共享的數據庫連接、增強安全性、身份驗證、角色權限等等，會非常的實用。

查詢參數和數據校驗

FastAPI 支持我們進行更加智能的數據校驗，比如一個字符串，我們希望用戶在傳遞的時候只能傳遞長度為 6 到 15 的字符串該怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()

@app.get("/user")
async def check_length(
        # 默認值為 None，應該聲明為 Optional[str]，當然聲明 str 也是可以的。只不過聲明為 str，那么默認值應該也是 str
        # 所以如果一個類型允許為空，那么更規范的做法應該是聲明為 Optional[類型]。
        password: Optional[str] = Query(None, min_length=6, max_length=15)
):
    return {"password": password}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

password 是可選的，但是一旦傳遞則必須傳遞字符串、而且還是長度在 6 到 15 之間的字符串。所以如果傳遞的是 None，那么在聲明默認值的時候 None 和 Query(None) 是等價的，只不過 Query 還支持其它的參數來對參數進行限制。

Query 里面除了限制最小長度和最大長度，還有其它的功能：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()


@app.get("/user")
async def check_length(
        password: str = Query("satori", min_length=6, max_length=15, regex=r"^satori")
):
    """此時的 password 默認值為 'satori'，並且傳遞的時候必須要以 'satori' 開頭
       但是值得注意的是 password 后面的是 str，不再是 Optional[str]，因為默認值不是 None 了
       當然這里即使寫成 Optional[str] 也是沒有什么影響的
    """
    return {"password": password}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

聲明為必須參數

我們通過 Query 可以限制參數的長度，但是問題來了，這個時候我還希望這個參數是必傳的該怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()

@app.get("/user")
async def check_length(
        password: str = Query(..., min_length=6)
):
    """將第一個參數換成 ... 即可實現該參數是必傳參數
    """
    return {"password": password}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

... 是 Python 中的一個特殊的對象，可以了解一下，通過它可以實現該參數是必傳參數。

查詢參數變成一個列表

如果我們指定了 a=1&a=2，那么我們在獲取 a 的時候如何才能得到一個列表呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()

@app.get("/items")
async def read_items(
        a1: str = Query(...),
        a2: List[str] = Query(...),
        b: List[str] = Query(...)
):
    return {"a1": a1, "a2": a2, "b": b}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

我們訪問一下，看看結果：

首先 "a2" 和 "b" 都是對應列表，然后 "a1" 只獲取了最后一個值。另外可能有人覺得我們這樣有點啰嗦，在函數聲明中可不可以這樣寫呢？

@app.get("/items")
async def read_items(
        a1: str,
        a2: List[str],
        b: List[str]
):
    return {"a1": a1, "a2": a2, "b": b}

對於 a1 是可以的，但是 a2 和 b 不行。對於類型為 list 的查詢參數，無論有沒有默認值，你都必須要顯式的加上 Query 來表示必傳參數。如果允許為 None（或者有默認值）的話，那么應該這么寫：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()


@app.get("/items")
async def read_items(
        a1: str,
        a2: Optional[List[str]] = Query(None),
        b: List[str] = Query(["1", "嘿嘿"])
):
    return {"a1": a1, "a2": a2, "b": b}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

給參數起別名

問題來了，假設我們定義的查詢參數名叫 item-query，那么由於它要體現在函數參數中、而這顯然不符合 Python 變量的命名規范，這個時候要怎么做呢？答案是起一個別名。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()

@app.get("/items")
async def read_items(
        # 通過 url 的時候使用別名即可
        item1: Optional[str] = Query(None, alias="item-query"),
        item2: str = Query("哈哈", alias="@@@@"),
        item3: str = Query(..., alias="$$$$")  # item3 是必傳的
):
    return {"item1": item1, "item2": item2, "item3": item3}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

數值檢測

Query 不僅僅支持對字符串的校驗，還支持對數值的校驗，里面可以傳遞 gt、ge、lt、le 這幾個參數，相信這幾個參數不用說你也知道是干什么的，我們舉例說明：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()

@app.get("/items")
async def read_items(
        # item1 必須大於 5
        item1: int = Query(..., gt=5),
        # item2 必須小於等於 7
        item2: int = Query(..., le=7),
        # item3 必須必須等於 10
        item3: int = Query(..., ge=10, le=10)
):
    return {"item1": item1, "item2": item2, "item3": item3}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

Query 還是比較強大了，當然內部還有一些其它的參數是針對 docs 交互文檔的，有興趣可以自己了解一下。

路徑參數和數據校驗

查詢參數數據校驗使用的是 Query，路徑參數數據校驗使用的是 Path，兩者的使用方式一模一樣，沒有任何區別。

from fastapi import FastAPI, Path
import uvicorn

app = FastAPI()

@app.get("/items/{item-id}")
async def read_items(item_id: int = Path(..., alias="item-id")):
    return {"item_id": item_id}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

因為路徑參數是必須的，它是路徑的一部分，所以我們應該使用 ... 將其標記為必傳參數。當然即使不這么做也無所謂，因為指定了默認值也用不上，因為路徑參數不指定壓根就匹配不到相應的路由。至於一些其它的校驗，和查詢參數一模一樣，所以這里不再贅述了。

不過我們之前說過，路徑參數應該在查詢參數的前面，盡管 FastAPI 沒有這個要求，但是這樣寫明顯更舒服一些。但是問題來了，如果路徑參數需要指定別名，但是某一個查詢參數不需要，這個時候就會出現問題：

@app.get("/items/{item-id}")
async def read_items(q: str,
                     item_id: int = Path(..., alias="item-id")):

    return {"item_id": item_id, "q": q}

顯然此時 Python 的語法就決定了 item_id 就必須放在 q 的后面，當然這么做是完全沒有問題的，FastAPI 對參數的先后順序沒有任何要求，因為它是通過參數的名稱、類型和默認值聲明來檢測參數，而不在乎參數的順序。但此時我們就要讓 item_id 在 q 的前面要怎么做呢？

@app.get("/items/{item-id}")
async def read_items(*, item_id: int = Path(..., alias="item-id"),
                     q: str):
    
    return {"item_id": item_id, "q": q}

此時就沒有問題了，通過將第一個參數設置為 *，使得 item_id 和 q 都必須通過關鍵字傳遞，所以此時默認參數在非默認參數之前也是允許的。當然我們也不需要擔心 FastAPI 傳參的問題，你可以認為它所有的參數都是通過關鍵字參數的方式傳遞的。

Request

Request 是什么？首先我們知道任何一個請求都對應一個 Request 對象，請求的所有信息都在這個 Request 對象中，FastAPI 也不例外。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Request
import uvicorn

app = FastAPI()


@app.get("/girl/{user_id}")
async def read_girl(user_id: str,
                    request: Request):
    """路徑參數是必須要體現在參數中，但是查詢參數可以不寫了
       因為我們定義了 request: Request，那么請求相關的所有信息都會進入到這個 Request 對象中"""
    header = request.headers  # 請求頭
    method = request.method  # 請求方法
    cookies = request.cookies  # cookies
    query_params = request.query_params  # 查詢參數
    return {"name": query_params.get("name"), "age": query_params.get("age"), "hobby": query_params.getlist("hobby")}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

我們通過 Request 對象可以獲取所有請求相關的信息，我們之前當參數傳遞不對的時候，FastAPI 會自動幫我們返回錯誤信息，但通過 Request 我們就可以自己進行解析、自己指定返回的錯誤信息了。

Response

既然有 Request，那么必然會有 Response，盡管我們可以直接返回一個字典，但 FastAPI 實際上會幫我們轉成一個 Response 對象。

Response 內部接收如下參數：

• content：返回的數據
• status_code：狀態碼
• headers：返回的請求頭
• media_type：響應類型(就是 HTML 中 Content-Type，只不過這里換了個名字)
• background：接收一個任務，Response 在返回之后會自動異步執行（這里不做介紹，后面會說）

舉個栗子：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Request, Response
import uvicorn
import orjson

app = FastAPI()


@app.get("/girl/{user_id}")
async def read_girl(user_id: str,
                    request: Request):
    query_params = request.query_params  # 查詢參數
    data = {"name": query_params.get("name"), "age": query_params.get("age"), "hobby": query_params.getlist("hobby")}
    # 實例化一個 Response 對象
    response = Response(
        # content，我們需要手動轉成 json 字符串，如果直接返回字典的話，那么在包裝成 Response 對象的時候會自動幫你轉
        orjson.dumps(data),
        # status_code，狀態碼
        201,
        # headers，響應頭
        {"Token": "xxx"},
        # media_type，就是 HTML 中的 Content-Type
        "application/json",  
    )
    # 如果想設置 cookie 的話，那么通過 response.set_cookie 即可
    # 刪除 cookie 則是 response.delete_cookie
    return response


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

通過 Response 我們可以實現請求頭、狀態碼、cookie 等自定義。

另外除了 Response 之外還有很多其它類型的響應，它們都在 fastapi.responses 中，比如：FileResponse、HTMLResponse、PlainTextResponse 等等。它們都繼承了 Response，只不過會自動幫你設置響應類型，舉個栗子：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.responses import Response, HTMLResponse
import uvicorn

app = FastAPI()


@app.get("/index")
async def index():

    response1 = HTMLResponse("<h1>你好呀</h1>")
    response2 = Response("<h1>你好呀</h1>", media_type="text/html")
    # 以上兩者是等價的，在 HTMLResponse 中會自動將 media_type 設置成 text/html
    return response1


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

其它類型的請求與響應

FastAPI 除了 GET 請求之外，還支持其它類型，比如：POST、PUT、DELETE、OPTIONS、HEAD、PATCH、TRACE 等等。而常見的也就 GET、POST、PUT、DELETE，介紹完了 GET，我們來說一說其它類型的請求。

顯然對應 POST、PUT 等類型的請求，我們必須要能夠解析出請求體，並且能夠構造出響應體。

Model

在 FastAPI 中，請求體和響應體都對應一個 Model，舉個栗子：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicorn

app = FastAPI()


class Girl(BaseModel):
    """數據驗證是通過 pydantic 實現的，我們需要從中導入 BaseModel，然后繼承它"""
    name: str
    age: Optional[str] = None
    length: float
    hobby: List[str]  # 對於 Model 中的 List[str] 我們不需要指定 Query（准確的說是 Field）


@app.post("/girl")
async def read_girl(girl: Girl):
    # girl 就是我們接收的請求體，它需要通過 json 來傳遞，並且這個 json 要有上面的四個字段（age 可以沒有）
    # 通過 girl.xxx 的方式我們可以獲取和修改內部的所有屬性
    return dict(girl)  # 直接返回 Model 對象也是可以的


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

除了使用這種方式之外，我們還可以使用之前說的 Request 對象：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Request
import uvicorn

app = FastAPI()

@app.post("/girl")
async def read_girl(request: Request):
    # 是一個協程，所以需要 await
    data = await request.body()
    print(data)

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

首先我們在使用 requests 模塊發送 post 請求的時候可以通過 data 參數傳遞、也可以通過 json 參數。

當通過 json={"name": "satori", "age": 16, "length": 155.5} 傳遞的時候，會將其轉成 json 字符串進行傳輸，程序中的 print 打印如下：

b'{"name": "satori", "age": 16, "length": 155.5}'

如果我們是用過 data 參數發請求的話（值不變），那么會將其拼接成 k1=v1&k2=v2 的形式再進行傳輸（相當於表單提交，后面說），程序中打印如下：

b'name=satori&age=16&length=155.5'

所以我們看到 await request.body() 得到的就是最原始的字節流，而除了 await request.body() 之外還有一個 await request.json()，只是后者在內部在調用了前者拿到字節流之后、自動幫你 loads 成了字典。因此使用 await request.json() 也側面要求，我們必須在發送請求的時候必須使用 json 參數傳遞（傳遞的是字典轉成的 json、所以也能解析成字典），否則使用 await request.json() 是無法正確解析的。

路徑參數、查詢參數、請求體

我們可以將三者混合在一起：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicorn

app = FastAPI()


class Girl(BaseModel):
    name: str
    age: Optional[str] = None
    length: float
    hobby: List[str]


@app.post("/girl/{user_id}")
async def read_girl(user_id,
                    q: str,
                    girl: Girl):

    return {"user_id": user_id, "q": q, **dict(girl)}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

里面我們同時指定了路徑參數、查詢參數和請求體，FastAPI 依然是可以正確區分的，當然我們也可以使用 Request 對象。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List, Dict
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicorn

app = FastAPI()

@app.post("/girl/{user_id}")
async def read_girl(user_id,
                    request: Request):

    q = request.query_params.get("q")
    data: Dict = await request.json()
    data.update({"user_id": user_id, "q": q})
    return data


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

可以自己測試一下，仍然是可以正確返回的。

多個請求體參數

我們上面的只接收一個 json 請求體，如果是接收兩個呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicorn

app = FastAPI()


class Girl(BaseModel):
    name: str
    age: Optional[str] = None


class Boy(BaseModel):
    name: str
    age: int


@app.post("/boy_and_girl")
async def read_boy_and_girl(girl: Girl,
                            boy: Boy):
    return {"girl": dict(girl), "boy": dict(boy)}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

此時在傳遞的時候，應該按照如下方式傳遞：

應該將兩個 json 嵌套在一起，組成一個更大的 json，至於 key 就是我們的函數參數名。因此這種方式其實就等價於：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List, Dict
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicorn

app = FastAPI()

class BoyAndGirl(BaseModel):
    girl: Dict
    boy: Dict

@app.post("/boy_and_girl")
async def read_boy_and_girl(boy_and_girl: BoyAndGirl):
    return dict(boy_and_girl)

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

這種方式也是可以實現的，只不過就字典內部的字典的不可進行限制了。當然啦，我們仍然可以使用 Request 對象，得到字典之后自己再進行判斷，因為對於 json 而言，內部的字段可能是會變的，而且最關鍵的是字段可能非常多。這個時候，我個人更傾向於使用 Request 對象。

Form 表單

我們調用 requests.post，如果參數通過 data 傳遞的話，則相當於提交了一個 form 表單，那么在 FastAPI 中可以通過 await request.form() 進行獲取，注意：內部同樣是先調用 await request.body()。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Request, Response
import uvicorn

app = FastAPI()

@app.post("/girl")
async def girl(request: Request):
    # 此時 await request.json() 報錯，因為是通過 data 參數傳遞的，相當於 form 表單提交
    # 如果是通過 json 參數傳遞，那么 await request.form() 會得到一個空表單
    form = await request.form()
    return [form.get("name"), form.getlist("age")]

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

當然我們也可以通過其它方式：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Form
import uvicorn

app = FastAPI()

@app.post("/user")
async def get_user(username: str = Form(...),
                   password: str = Form(...)):
    return {"username": username, "password": password}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

像 Form 表單，查詢參數、路徑參數等等，都可以和 Request 對象一起使用，像上面的例子，如果我們多定義一個 request: Request，那么我們仍然可以通過 await request.form() 拿到相關的表單信息。所以如果你覺得某個參數不適合類型注解，那么你可以單獨通過 Request 對象進行解析。

文件上傳

那么問題來了，FastAPI 如何接收用戶的文件上傳呢？首先如果想使用文件上傳功能，那么你必須要安裝一個包 python-multipart，直接 pip install python-multipart 即可。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, File, UploadFile
import uvicorn

app = FastAPI()

@app.post("/file1")
async def file1(file: bytes = File(...)):
    return f"文件長度: {len(file)}"

@app.post("/file2")
async def file1(file: UploadFile = File(...)):
    return f"文件名: {file.filename}, 文件大小: {len(await file.read())}"


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

我們看到一個直接獲取字節流，另一個是獲取類似於文件句柄的對象。如果是多個文件上傳要怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import List
from fastapi import FastAPI, UploadFile, File
import uvicorn

app = FastAPI()

@app.post("/file")
async def file(files: List[UploadFile] = File(...)):
    """指定類型為列表即可"""
    for idx, f in enumerate(files):
        files[idx] = f"文件名: {f.filename}, 文件大小: {len(await f.read())}"
    return files

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

此時我們就實現了 FastAPI 文件上傳，當然文件上傳並不影響我們處理表單，可以自己試一下同時處理文件和表單。

返回靜態資源

下面來看看 FastAPI 如何返回靜態資源，首先我們需要安裝 aiofiles，直接 pip 安裝即可。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
import uvicorn

app = FastAPI()

# name 參數只是起一個名字，FastAPI 內部使用
app.mount("/static", StaticFiles(directory=r"C:\Users\satori\Desktop\bg"), name="static")

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

瀏覽器輸入：localhost:5555/static/1.png，那么會返回 C:\Users\satori\Desktop\bg 下的 1.png 文件。

錯誤處理

錯誤處理也是一個不可忽視的點，錯誤有很多種，比如：

• 客戶端沒有足夠的權限執行此操作
• 客戶端沒有訪問某個資源的權限
• 客戶端嘗試訪問一個不存在的資源
• ...

這個時候我們應該將錯誤通知相應的客戶端，這個客戶端可以瀏覽器、代碼程序、IoT 設備等等。

但是就我個人而言，更傾向於使用 Response 對象，將里面的  status_code 設置為 404，然后在返回的 json 中指定錯誤信息。不過 FastAPI 內部也提供了一些異常類：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, HTTPException
import uvicorn

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id != "foo":
        # 里面還可以傳入 headers 設置響應頭
        raise HTTPException(status_code=404, detail="item 沒有發現")
    return {"item": "bar"}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

HTTPException 是一個普通的 Python 異常類（繼承了 Exception），它攜帶了 API 的相關信息，既然是異常，那么我們不能 return、而是要 raise。

個人覺得這個不是很常用，至少我本人很少用這種方式返回錯誤，因為它能夠攜帶的信息太少了。

自定義異常

FastAPI 內部提供了一個 HTTPException，但是我們也可以自定義，但是注意：我們自定義完異常之后，還要定義一個 handler，將異常和 handler 綁定在一起，然后引發該異常的時候就會觸發相應的 handler。

from fastapi import FastAPI, Request
from fastapi.responses import ORJSONResponse
import uvicorn

app = FastAPI()


class ASCIIException(Exception):
    """何もしません"""

# 通過裝飾器的方式，將 ASCIIException 和 ascii_exception_handler 綁定在一起
@app.exception_handler(ASCIIException)
async def ascii_exception_handler(request: Request, exc: ASCIIException):
    """當引發 ASCIIException 的時候，會觸發 ascii_exception_handler 的執行
       同時會將 request 和 exception 傳過去"""
    return ORJSONResponse(status_code=404, content={"code": 404, "message": "你必須傳遞 ascii 字符串"})

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if not item_id.isascii():
        raise ASCIIException
    return {"item": f"get {item_id}"}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

關於 Request、Response，我們除了可以通過 fastapi 進行導入，還可以通過 starlette 進行導入，因為 fastapi 的路由映射是通過 starlette 來實現的。當然我們直接從 fastapi 中進行導入即可。

自定義 404

當訪問一個不存在的 URL，我們應該提示用戶，比如：您要找到頁面去火星了。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
from fastapi.exceptions import StarletteHTTPException
import uvicorn

app = FastAPI()

@app.exception_handler(StarletteHTTPException)
async def not_found(request, exc):
    return ORJSONResponse({"code": 404, "message": "您要找的頁面去火星了。。。"})

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

此時當我們訪問一個不存在的 URL 時，就會返回我們自定義的 JSON 字符串。

后台任務

如果一個請求耗時特別久，那么我們可以將其放在后台執行，而 FastAPI 已經幫我們做好了這一步。我們來看一下：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
import time
from fastapi import FastAPI, BackgroundTasks
import uvicorn

app = FastAPI()

def send_email(email: str, message: str = ""):
    """發送郵件，假設耗時三秒"""
    time.sleep(3)
    print(f"三秒之后郵件發送給 {email!r}, 郵件信息: {message!r}")

@app.get("/user/{email}")
async def order(email: str, bg_tasks: BackgroundTasks):
    """這里需要多定義一個參數
       此時任務就被添加到后台，當 Response 對象返回之后觸發"""
    bg_tasks.add_task(send_email, email, message="這是一封郵件")
    # 我們在之前介紹 Response 的時候說過，里面有一個參數 background
    # 所以我們也可以將任務放在那里面
    # 因此我們還可以：
    # return Response(orjson.dumps({"message": "郵件發送成功"}), background=BackgroundTask(send_email, email, message="這是一封郵件"))
    return {"message": "郵件發送成功"}


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

首先請求肯定是成功的：

然后響應的 3 秒，終端會出現如下打印：

所以此時任務是被后台執行了的，注意：任務是在響應返回之后才后台執行。

APIRouter

APIRouter 類似於 Flask 中的藍圖，可以更好的組織大型項目，舉個栗子：

在我當前的工程目錄中有一個 app 目錄和一個 main.py，其中 app 目錄中有一個 app01.py，然后我們看看它們是如何組織的。

# app/app01.py
from fastapi import APIRouter

router = APIRouter(prefix="/router")

# 以后訪問的時候要通過 /router/v1 來訪問
@router.get("/v1")
async def v1():
    return {"message": "hello world"}


# main.py
from fastapi import FastAPI
from app.app01 import router
import uvicorn

app = FastAPI()

# 將 router 注冊到 app 中，相當於 Flask 中的 register_blueprint
app.include_router(router)

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

然后可以在外界通過 /router/v1 的方式來訪問。

中間件

中間件在 web 開發中可以說是非常常見了，說白了中間件就是一個函數或者一個類。在請求進入視圖函數之前，會先經過中間件（被稱為請求中間件），而在中間件里面，我們可以對請求進行一些預處理，或者實現一個攔截器等等；同理當視圖函數返回響應之后，也會經過中間件（被稱為響應中間件），在中間件里面，我們也可以對響應進行一些潤色。

自定義中間件

在 FastAPI 里面也支持像 Flask 一樣自定義中間件，但是 Flask 里面有請求中間件和響應中間件，但是在 FastAPI 里面這兩者合二為一了，我們看一下用法。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Request, Response
import uvicorn
import orjson

app = FastAPI()


@app.get("/")
async def view_func(request: Request):
    return {"name": "古明地覺"}


@app.middleware("http")
async def middleware(request: Request, call_next):
    """
    定義一個協程函數，然后使用 @app.middleware("http") 裝飾，即可得到中間件
    """
    # 請求到來時會先經過這里的中間件
    if request.headers.get("ping", "") != "pong":
        response = Response(content=orjson.dumps({"error": "請求頭中缺少指定字段"}),
                            media_type="application/json",
                            status_code=404)
        # 當請求頭中缺少 "ping": "pong"，在中間件這一步就直接返回了，就不會再往下走了
        # 所以此時就相當於實現了一個攔截器
        return response
    # 然后，如果條件滿足，則執行 await call_next(request)，關鍵是這里的 call_next
    # 如果該中間件后面還有中間件，那么 call_next 就是下一個中間件；如果沒有，那么 call_next 就是對應的視圖函數
    # 這里顯然是視圖函數，因此執行之后會拿到視圖函數返回的 Response 對象
    # 所以我們看到在 FastAPI 中，請求中間件和響應中間件合在一起了
    response: Response = await call_next(request)
    # 這里我們在設置一個響應頭
    response.headers["status"] = "success"
    return response


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

我們可以測試一下：

測試結果也印證了我們的結論。

內置的中間件

通過自定義中間件，我們可以在不修改視圖函數的情況下，實現功能的擴展。但是除了自定義中間件之外，FastAPI 還提供了很多內置的中間件。

app = FastAPI()

# 要求請求協議必須是 https 或者 wss，如果不是，則自動跳轉
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
app.add_middleware(HTTPSRedirectMiddleware)

# 請求中必須包含 Host 字段，為防止 HTTP 主機報頭攻擊，並且添加中間件的時候，還可以指定一個 allowed_hosts，那么它是干什么的呢？
# 假設我們有服務 a.example.com, b.example.com, c.example.com
# 但我們不希望用戶訪問 c.example.com，就可以像下面這么設置，如果指定為 ["*"]，或者不指定 allow_hosts，則表示無限制
from starlette.middleware.trustedhost import TrustedHostMiddleware
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["a.example.com", "b.example.com"])

# 如果用戶的請求頭的 Accept-Encoding 字段包含 gzip，那么 FastAPI 會使用 GZip 算法壓縮
# minimum_size=1000 表示當大小不超過 1000 字節的時候就不壓縮了
from starlette.middleware.gzip import GZipMiddleware
app.add_middleware(GZipMiddleware, minimum_size=1000)

除了這些，還有其它的一些內置的中間件，可以自己查看一下，不過不是很常用。

CORS

CORS 過於重要，我們需要單獨拿出來說。

CORS（跨域資源共享）是指瀏覽器中運行的前端里面擁有和后端通信的 JavaScript 代碼，而前端和后端處於不同源的情況。源：協議（http、https）、域（baidu.com、app.com、localhost）以及端口（80、443、8000），只要有一個不同，那么就是不同源。比如下面都是不同的源：

• http://localhost
• https://localhost
• http://localhost:8080

即使它們都是 localhost，但是它們使用了不同的協議或端口，所以它們是不同的源。假設你的前端運行在 localhost:8080，並且嘗試與 localhost:5555 進行通信；然后瀏覽器會向后端發送一個 HTTP OPTIONS 請求，后端會發送適當的 headers 來對這個源進行授權；所以后端必須有一個 "允許的源" 列表，如果前端對應的源是被允許的，瀏覽器才會允許前端向后端發請求，否則就會出現跨域失敗。

而默認情況下，前后端必須是在同一個源，如果不同源那么前端就會請求失敗。而前后端分離早已成為了主流，因此跨域問題是必須要解決的。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import uvicorn

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    # 允許跨域的源列表，例如 ["http://www.example.org"] 等等，["*"] 表示允許任何源
    allow_origins=["*"],
    # 跨域請求是否支持 cookie，默認是 False，如果為 True，allow_origins 必須為具體的源，不可以是 ["*"]
    allow_credentials=False,
    # 允許跨域請求的 HTTP 方法列表，默認是 ["GET"]
    allow_methods=["*"],
    # 允許跨域請求的 HTTP 請求頭列表，默認是 []，可以使用 ["*"] 表示允許所有的請求頭
    # 當然 Accept、Accept-Language、Content-Language 以及 Content-Type 總之被允許的
    allow_headers=["*"],
    # 可以被瀏覽器訪問的響應頭, 默認是 []，一般很少指定
    # expose_headers=["*"]
    # 設定瀏覽器緩存 CORS 響應的最長時間，單位是秒。默認為 600，一般也很少指定
    # max_age=1000
)

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

以上即可解決跨域問題。

高階操作

下面我們看一些 FastAPI 的高階操作，這些操作有的不一定能用上，但用上了確實會方便許多。

其它的響應

返回 json 數據可以是：JSONResponse、UJSONResponse、ORJSONResponse，Content-Type 是 application/json；返回 html 是 HTMLResponse，Content-Type 是 text/html；返回 PlainTextResponse，Content-Type 是 text/plain。但是我們還可以有三種響應，分別是返回重定向、字節流、文件。

重定向

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.responses import RedirectResponse
import uvicorn

app = FastAPI()


@app.get("/index")
async def index():
    return RedirectResponse("https://www.bilibili.com")


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

頁面中訪問 /index 會跳轉到 bilibili。

字節流

返回字節流需要使用異步生成器的方式：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import uvicorn

app = FastAPI()

async def some_video():
    for i in range(5):
        yield f"video {i} bytes ".encode("utf-8")

@app.get("/index")
async def index():
    return StreamingResponse(some_video())


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

如果有文件對象，那么也是可以直接返回的。

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import uvicorn

app = FastAPI()


@app.get("/index")
async def index():
    return StreamingResponse(open("main.py", encoding="utf-8"))


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

文件

返回文件的話，還可以通過 FileResponse：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.responses import FileResponse
import uvicorn

app = FastAPI()

@app.get("/index")
async def index():
    # filename 如果給出，它將包含在響應的 Content-Disposition 中。
    return FileResponse("main.py", filename="這不是main.py")

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

可以自己發請求測試一下。

HTTP 驗證

如果當用戶訪問某個請求的時候，我們希望其輸入用戶名和密碼來確認身份的話該怎么做呢？

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Depends
from fastapi.security import HTTPBasic, HTTPBasicCredentials
import uvicorn

app = FastAPI()

security = HTTPBasic()

@app.get("/index")
async def index(credentials: HTTPBasicCredentials = Depends(security)):
    return {"username": credentials.username, "password": credentials.password}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

當用戶訪問 /index 的時候，會提示輸入用戶名和密碼：

輸入完畢之后，信息會保存在 credentials，我們可以獲取出來進行驗證。

websocket

然后我們來看看 FastAPI 如何實現 websocket：

# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI
from fastapi.websockets import WebSocket
import uvicorn

app = FastAPI()

@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    while True:
        # websocket.receive_bytes()
        # websocket.receive_json()
        data = await websocket.receive_text()
        await websocket.send_text(f"收到來自客戶端的回復: {data}")

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5555)

然后我們通過瀏覽器進行通信：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
    <script>
        ws = new WebSocket("ws://localhost:5555/ws");
 
        //如果連接成功, 會打印下面這句話, 否則不會打印
        ws.onopen = function () {
            console.log('連接成功')
        };
 
        //接收數據, 服務端有數據過來, 會執行
        ws.onmessage = function (event) {
            console.log(event)
        };
 
        //服務端主動斷開連接, 會執行.
        //客戶端主動斷開的話, 不執行
        ws.onclose = function () {  }
 
    </script>
</body>
</html>

FastAPI 服務的部署

目前的話，算是介紹了 FastAPI 的絕大部分內容，然后我們來看看 FastAPI 服務的部署，其實部署很簡單，直接 uvicorn.run 即可。但是這里面有很多的參數，我們主要是想要介紹這些參數。

def run(app, **kwargs):
    config = Config(app, **kwargs)
    server = Server(config=config)
    ...
    ...

我們看到 app 和 **kwargs 都傳遞給了 Config，所以我們只需要看 Config 里面都有哪些參數即可。這里選出一部分：

• app：第一個參數，不需要解釋
• host：監聽的ip
• port：監聽的端口
• uds：綁定的 unix domain socket，一般不用
• fd：從指定的文件描述符中綁定 socket
• loop：事件循環實現，可選項為 auto|asyncio|uvloop|iocp
• http：HTTP 協議實現，可選項為 auto|h11|httptools
• ws：websocket 協議實現，可選項為 auto|none|websockets|wsproto
• lifespan：lifespan 實現，可選項為 auto|on|off
• env_file：環境變量配置文件
• log_config：日志配置文件
• log_level：日志等級
• access_log：是否記錄日志
• use_colors：是否帶顏色輸出日志信息
• interface：應用接口，可選 auto|asgi3|asgi2|wsgi
• debug：是否開啟 debug 模式
• reload：是否自動重啟
• reload_dirs：要自動重啟的目錄
• reload_delay：多少秒后自動重啟
• workers：工作進程數
• limit_concurrency：並發的最大數量
• limit_max_requests：能 hold 住的最大請求數

小結

總的來說，FastAPI 算是當前最流行的異步框架了，並且它完全可以在生產中使用，是值得信賴的。當然使用異步框架，最重要的是要搭配一個異步驅動去訪問數據庫，因為 web 服務的瓶頸都是在數據庫上面。