FastAPI 非同步程式設計：`async / await` 完全指南

簡介

在現代 Web 應用中，高併發與快速回應已成為基本需求。傳統的同步阻塞模型在面對大量 I/O（如資料庫、外部 API、檔案讀寫）時，容易形成「瓶頸」——每個請求必須等前一個請求完成才能繼續執行，導致資源利用率低落。

FastAPI 內建支援 非同步（asynchronous） 程式設計，藉由 Python 3.7+ 的 async / await 語法，讓開發者能以直觀的方式撰寫非阻塞的端點（endpoint）。掌握這套機制，不僅可以提升服務的吞吐量，還能減少硬體成本、提升使用者體驗。

本篇文章將從概念、實作範例、常見陷阱 以及 最佳實踐 四個面向，完整說明在 FastAPI 中如何正確使用 async / await，適合剛入門的初學者，也能為已有基礎的開發者提供實務參考。

核心概念

1. 同步 vs 非同步

項目	同步 (Blocking)	非同步 (Non‑blocking)
執行流程	每個任務必須等前一個完成	任務可在等待 I/O 時釋放執行緒
資源使用	需要大量執行緒或進程	只需要少量執行緒（通常 1 個 event loop）
程式碼可讀性	直線式，易懂	需要 `async` / `await`，稍有學習曲線

在 FastAPI 中，只要把路由處理函式宣告為 async def，框架會自動將其交給 asyncio event loop 處理，讓 I/O 操作（例如 await httpx.get(...)）在等待回應時不會阻塞其他請求。

2. 為什麼要在 FastAPI 中使用 `async`

效能提升：對於大量 I/O 密集型的服務（如呼叫外部 API、讀寫資料庫），async 能讓單個實例同時處理更多請求。
資源節省：不需要為每個請求建立獨立的執行緒或進程，減少記憶體與 CPU 開銷。
與現代 Python 生態相容：許多第三方套件（如 httpx, databases, aioredis）已提供非同步介面，直接搭配 FastAPI 使用最為順手。

3. `async` / `await` 的基本語法

# 同步函式
def get_user_sync(user_id: int) -> dict:
    user = db.query(User).filter(User.id == user_id).first()
    return {"id": user.id, "name": user.name}

# 非同步函式
async def get_user_async(user_id: int) -> dict:
    user = await db.fetch_one("SELECT * FROM users WHERE id = :id", {"id": user_id})
    return {"id": user["id"], "name": user["name"]}

async def：宣告一個協程（coroutine），此函式會回傳 coroutine 物件，需在 await 或事件迴圈中執行。
await：等待一個可 await 的物件（如 asyncio.sleep, 非同步資料庫查詢），在等待期間會把控制權交還給 event loop。

⚠️ 重點：只能在 async def 裡使用 await，而且只能 await 支援非同步 的函式或物件。

程式碼範例

以下示範 5 個在 FastAPI 中常見的 async / await 使用情境，並說明每行程式的目的。

範例 1：簡單的非同步端點

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/ping")
async def ping():
    """一個最簡單的非同步回應，模擬 0.5 秒的延遲"""
    await asyncio.sleep(0.5)   # 非阻塞等待
    return {"msg": "pong"}

await asyncio.sleep(0.5) 會讓 event loop 先處理其他請求，而不是卡住整個執行緒。

範例 2：呼叫外部 API（使用 httpx）

import httpx
from fastapi import FastAPI

app = FastAPI()
client = httpx.AsyncClient()   # 建立全域的非同步 http 客戶端

@app.get("/weather/{city}")
async def get_weather(city: str):
    """向第三方天氣 API 發送非同步請求"""
    url = f"https://api.example.com/weather/{city}"
    resp = await client.get(url)        # 非同步 HTTP GET
    data = resp.json()
    return {"city": city, "temp": data["temp_celsius"]}

使用 httpx.AsyncClient 可以在 同一個事件迴圈 中同時發出多筆 HTTP 請求，避免阻塞。

範例 3：非同步資料庫查詢（使用 databases）

import databases
from fastapi import FastAPI

DATABASE_URL = "postgresql+asyncpg://user:pwd@localhost/dbname"
database = databases.Database(DATABASE_URL)

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()   # 啟動時連線資料庫

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    query = "SELECT id, name, email FROM users WHERE id = :uid"
    row = await database.fetch_one(query=query, values={"uid": user_id})
    if row:
        return dict(row)
    return {"error": "User not found"}

databases 套件提供完整的非同步 CRUD API，與 await 搭配即可保持端點的非阻塞特性。

範例 4：同時執行多個非同步任務（`asyncio.gather`）

import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()
client = httpx.AsyncClient()

async def fetch_price(symbol: str) -> float:
    resp = await client.get(f"https://api.example.com/price/{symbol}")
    return resp.json()["price"]

@app.get("/prices")
async def get_prices():
    """一次取得多支股票的即時價格，使用 asyncio.gather 並行執行"""
    symbols = ["AAPL", "GOOG", "TSLA", "MSFT"]
    tasks = [fetch_price(sym) for sym in symbols]
    prices = await asyncio.gather(*tasks)   # 並行等待所有任務完成
    return {sym: price for sym, price in zip(symbols, prices)}

asyncio.gather 能將多筆 I/O 任務同時排入 event loop，顯著縮短總耗時。

範例 5：使用背景任務（BackgroundTasks）處理非即時工作

from fastapi import FastAPI, BackgroundTasks
import asyncio

app = FastAPI()

async def write_log(message: str):
    await asyncio.sleep(0.1)          # 模擬磁碟 I/O
    with open("access.log", "a", encoding="utf-8") as f:
        f.write(message + "\n")

@app.post("/items/")
async def create_item(name: str, background_tasks: BackgroundTasks):
    """建立新項目，同時把寫入 log 的工作交給背景任務"""
    # 假設此處有資料庫寫入的非同步程式...
    background_tasks.add_task(write_log, f"Item created: {name}")
    return {"msg": f"Item '{name}' created"}

BackgroundTasks 允許在回應送出後，仍能在同一個事件迴圈中執行輕量級的非同步工作，避免阻塞主請求流程。

常見陷阱與最佳實踐

陷阱	說明	解決方案
在同步函式中使用 `await`	`await` 只能出現在 `async def` 裡，否則會拋出 `SyntaxError`。	把函式改為 `async def`，或改用非同步函式的同步封裝（如 `asyncio.run`）
阻塞 I/O 混入非同步端點	若在 `async` 端點裡直接呼叫阻塞的函式（如 `requests.get`、`time.sleep`），會阻塞整個 event loop。	改用非同步版套件（`httpx.AsyncClient`、`asyncio.sleep`）或使用執行緒池 (`run_in_executor`)
過度使用 `async`	並非所有情況都需要非同步；若端點僅執行 CPU 密集運算，`async` 反而無法提升效能。	針對 I/O 密集型任務使用 `async`，CPU 密集型任務可考慮多程序或背景工作
未關閉非同步資源	如 `httpx.AsyncClient`、資料庫連線若未在應用關閉時釋放，會導致資源泄漏。	使用 `@app.on_event("shutdown")` 釋放資源，或在 `with` 內使用 `AsyncClient`
忘記 `await`	直接呼叫協程而未加 `await`，會得到 `coroutine` 物件而非執行結果，導致錯誤或未預期的行為。	確認每個非同步呼叫都有 `await`，或使用 `asyncio.create_task` 交由背景執行

最佳實踐小結

只在需要的地方使用 async：先確認是否有 I/O 操作，再決定是否改寫為非同步。
使用支援非同步的套件：如 httpx, databases, aioredis, asyncpg。
統一管理非同步資源：在 startup / shutdown 事件中建立與關閉連線，避免重複建立導致效能下降。
盡量避免混用阻塞與非阻塞程式碼：若必須呼叫阻塞函式，請使用 await asyncio.to_thread(blocking_func, ...) 或 loop.run_in_executor。
測試與監控：使用 locust、hey 等壓測工具觀察在高併發下的回應時間與資源使用，確保非同步改寫真的提升效能。

實際應用場景

場景	為何適合使用 `async / await`
呼叫多個第三方 API（如匯率、天氣、付款）	`asyncio.gather` 可同時發送請求，將總等待時間從 N × latency 降至 max(latency)。
即時聊天或推播服務	使用 WebSocket（FastAPI 原生支援）時，需長時間保持連線，非同步能避免每個連線佔用獨立執行緒。
大量資料讀寫（如日誌、檔案上傳）	非同步檔案 I/O（`aiofiles`）讓大量上傳不會阻塞其他請求。
背景任務與排程（如寄送 Email、產生報表）	`BackgroundTasks` 或 `Celery` 搭配 `async` 可在同一個服務內處理輕量任務，降低外部排程成本。
微服務間的同步呼叫	服務 A 呼叫服務 B、C、D 時，使用 `httpx.AsyncClient` 同時發起三筆請求，縮短總回應時間。

總結

async / await 是 Python 原生的非同步機制，在 FastAPI 中只要把端點寫成 async def，就能自動受益於事件迴圈的高效能調度。
正確使用非同步程式碼可以顯著提升 I/O 密集型服務的吞吐量與回應速度，同時減少硬體資源的需求。
開發時務必要避免 阻塞 I/O、未關閉資源 與 忘記 await 等常見陷阱，並遵循「只在需要的地方使用 async」的原則。
透過 httpx.AsyncClient、databases、asyncio.gather、BackgroundTasks 等工具，我們可以輕鬆構建 高併發、低延遲 的 API 服務。

掌握了上述概念與實務技巧，你就能在 FastAPI 專案中自信地運用 async / await，打造出既快速又 可擴展 的現代化 Web 應用。祝開發順利！

FastAPI 非同步程式設計：async / await 完全指南

簡介