FastAPI 與外部服務整合：呼叫外部 API（httpx / aiohttp）

---

簡介

在現代的微服務架構或前後端分離的應用中，FastAPI 常常需要與其他服務互相呼叫（例如第三方支付、天氣資訊、機器學習模型等）。如果不熟悉非同步 HTTP 客戶端的使用方式，往往會因為阻塞 I/O、錯誤處理不完整或資源浪費而造成效能瓶頸。

本篇文章將以 httpx 與 aiohttp 為例，說明在 FastAPI 中如何安全、有效率且易於維護地呼叫外部 API，並提供完整的程式碼範例、常見陷阱與最佳實踐，讓你從初學者快速晉升為中級開發者。

---

核心概念

1. 為什麼要使用非同步 HTTP 客戶端？

FastAPI 本身是基於 Starlette 的非同步框架，所有路由處理函式（async def）在執行期間若遇到 阻塞的 I/O 操作（例如 requests.get()），整個工作執行緒會被占用，導致 同時只能處理少量請求。使用支援 async/await 的 HTTP 客戶端（如 httpx、aiohttp）可以把等待時間交給事件迴圈，讓其他請求得以同時執行，提升整體吞吐量。

2. httpx 與 aiohttp 的差異

特性	httpx	aiohttp
API 風格	與 requests 相似，學習曲線低	更貼近底層的 asyncio，彈性較高
內建 HTTP/2 支援	✅（需要額外安裝 httpx[http2]）	❌
連線池管理	自動管理，支援 Client 物件	需要自行建立 ClientSession
文件與社群	文件完整、示例多	文件較簡潔，社群較小
使用情境	快速開發、需要 HTTP/2 時	需要高度自訂（如 WebSocket）時

兩者皆能滿足大多數 FastAPI 的外部呼叫需求，選擇哪一個主要取決於 團隊習慣與功能需求。

---

程式碼範例

以下範例全部採用 FastAPI 2.x（支援 async 路由），示範如何在不同情境下使用 httpx 與 aiohttp。

1️⃣ 基本的 httpx 非同步 GET 呼叫

# file: app/main.py
from fastapi import FastAPI, HTTPException
import httpx

app = FastAPI()

@app.get("/weather/{city}")
async def get_weather(city: str):
    """
    取得指定城市的即時天氣資訊（使用 httpx）。
    """
    url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid=YOUR_API_KEY"
    async with httpx.AsyncClient(timeout=10.0) as client:
        try:
            response = await client.get(url)
            response.raise_for_status()
        except httpx.HTTPError as exc:
            raise HTTPException(status_code=502, detail=f"外部服務錯誤: {exc}") from exc

    data = response.json()
    # 只回傳我們關心的欄位
    return {
        "city": data["name"],
        "temperature": data["main"]["temp"],
        "description": data["weather"][0]["description"]
    }

重點：

• 使用 httpx.AsyncClient 建立在每個請求內的臨時連線池，確保請求結束後自動關閉。
• raise_for_status() 會把 4xx/5xx 轉成例外，方便統一錯誤處理。

2️⃣ 共享 httpx 客戶端（提升效能）

在高頻呼叫的服務中，建立一次全域的 AsyncClient 可以避免每次請求都重新建立 TCP 連線。

# file: app/dependencies.py
import httpx
from fastapi import Depends

# 程式啟動時建立一次，整個應用共用
shared_client = httpx.AsyncClient(timeout=10.0, limits=httpx.Limits(max_connections=100, max_keepalive=20))

async def get_http_client() -> httpx.AsyncClient:
    return shared_client

# file: app/main.py
from fastapi import FastAPI, Depends, HTTPException
from .dependencies import get_http_client
import httpx

app = FastAPI()

@app.get("/exchange-rate")
async def get_exchange_rate(client: httpx.AsyncClient = Depends(get_http_client)):
    """
    取得美元對台幣匯率（使用共享 client）。
    """
    url = "https://api.exchangerate.host/latest?base=USD&symbols=TWD"
    try:
        r = await client.get(url)
        r.raise_for_status()
    except httpx.HTTPError as exc:
        raise HTTPException(status_code=502, detail=str(exc))

    return r.json()["rates"]

提示：

• 記得在 應用關閉時 呼叫 await shared_client.aclose()，可在 startup / shutdown 事件中完成。

3️⃣ 使用 aiohttp 呼叫 POST 並傳送 JSON

# file: app/main.py
from fastapi import FastAPI, HTTPException
import aiohttp
import asyncio

app = FastAPI()

@app.post("/translate")
async def translate_text(text: str, target_lang: str = "en"):
    """
    呼叫第三方翻譯 API（使用 aiohttp）。
    """
    api_url = "https://api.example.com/v1/translate"
    payload = {"text": text, "target_lang": target_lang}
    headers = {"Authorization": "Bearer YOUR_TOKEN", "Content-Type": "application/json"}

    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=8)) as session:
        try:
            async with session.post(api_url, json=payload, headers=headers) as resp:
                if resp.status != 200:
                    body = await resp.text()
                    raise HTTPException(status_code=502, detail=f"外部 API 錯誤 {resp.status}: {body}")
                result = await resp.json()
        except aiohttp.ClientError as exc:
            raise HTTPException(status_code=502, detail=f"連線失敗: {exc}") from exc

    return {"translated": result["translation"]}

關鍵：

• aiohttp.ClientSession 也是 連線池，建議在 較大的作用域（如 startup）建立一次，避免每次請求重建。

4️⃣ 並行呼叫多個外部服務（asyncio.gather）

# file: app/main.py
from fastapi import FastAPI, HTTPException
import httpx
import asyncio

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

async def fetch_weather(city: str):
    url = f"https://api.weather.com/v3/weather/conditions?city={city}&apiKey=KEY"
    r = await client.get(url)
    r.raise_for_status()
    return r.json()

async def fetch_news(topic: str):
    url = f"https://newsapi.org/v2/everything?q={topic}&apiKey=KEY"
    r = await client.get(url)
    r.raise_for_status()
    return r.json()["articles"][:3]

@app.get("/dashboard/{city}/{topic}")
async def dashboard(city: str, topic: str):
    """
    同時取得天氣與新聞，利用 asyncio.gather 並行執行。
    """
    try:
        weather, news = await asyncio.gather(
            fetch_weather(city),
            fetch_news(topic),
        )
    except httpx.HTTPError as exc:
        raise HTTPException(status_code=502, detail=str(exc))

    return {"city": city, "weather": weather, "news": news}

說明：

• asyncio.gather 會同時發起兩個 HTTP 請求，等待最慢者回傳，大幅降低總延遲。
• 若其中一個失敗，gather 會拋出例外，需在外層捕捉。

5️⃣ 設定重試機制（httpx + tenacity）

# file: app/utils.py
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import httpx

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(httpx.RequestError),
)
async def get_with_retry(client: httpx.AsyncClient, url: str) -> httpx.Response:
    """自動重試的 GET 請求，適用於不穩定的第三方服務。"""
    response = await client.get(url)
    response.raise_for_status()
    return response

# file: app/main.py
from fastapi import FastAPI, HTTPException
from .utils import get_with_retry
from .dependencies import shared_client

app = FastAPI()

@app.get("/unstable")
async def call_unstable_service():
    url = "https://unstable.example.com/api"
    try:
        r = await get_with_retry(shared_client, url)
    except httpx.HTTPError as exc:
        raise HTTPException(status_code=502, detail=f"重試失敗: {exc}")

    return r.json()

要點：

• 使用 tenacity 可以把重試邏輯抽離，保持路由函式的簡潔。
• 設定指數退避（exponential backoff）避免瞬間大量請求造成雪崩效應。

---

常見陷阱與最佳實踐

陷阱	描述	解決方案
阻塞 I/O	在 async def 中使用 requests、urllib 等同步庫。	改用 httpx.AsyncClient 或 aiohttp.ClientSession。
忘記關閉連線	AsyncClient/ClientSession 未在 shutdown 時關閉，導致資源洩漏。	在 @app.on_event("shutdown") 中呼叫 await client.aclose()。
過度建立 client	每個請求都 AsyncClient()，浪費 TCP 連線與 TLS 握手。	使用 共享 client 或 依賴注入（Depends）管理單例。
缺乏逾時設定	預設逾時過長，會卡住事件迴圈。	明確設定 timeout=，根據服務 SLA 調整。
未處理非 2xx 回應	直接返回 response.json()，若外部服務回傳 4xx/5xx 會產生錯誤。	使用 raise_for_status()，或自行檢查 response.status_code。
過度併發	同時發起大量請求，超過外部服務的速率限制。	在 httpx.Limits 或 aiohttp.TCPConnector 中設定 limit，加上 重試+退避。
缺少日誌	發生錯誤時難以追蹤根因。	使用 logging 記錄請求 URL、狀態碼、例外資訊。

最佳實踐清單

1. 統一錯誤模型：將所有外部 API 錯誤轉成 HTTPException（或自訂錯誤類別），讓前端能一致處理。
2. 依賴注入：利用 FastAPI 的 Depends 把 client 注入路由，便於測試與替換。
3. 環境變數：將 API 金鑰、URL、逾時等配置寫入 .env，使用 pydantic.Settings 管理。
4. 測試：使用 httpx.MockTransport 或 aioresponses 模擬外部回應，確保路由邏輯不受網路影響。
5. 安全：不要在程式碼中硬編碼金鑰，使用 Header Authorization 或 OAuth2 流程。
6. 監控：結合 Prometheus 或 OpenTelemetry 記錄外部呼叫的 latency、成功率等指標。

---

實際應用場景

場景	為何需要外部 API	建議使用方式
第三方支付	必須向金流平台送出交易請求、驗證回傳結果。	使用 共享 httpx client，加上 重試與簽章驗證。
即時天氣或地理資訊	前端顯示使用者所在位置的天氣、氣象警報。	使用 非同步 GET，配合 快取（Redis） 減少頻繁呼叫。
機器學習模型服務	將文字或影像上傳至模型微服務，取得預測結果。	使用 aiohttp 的 multipart/form-data 上傳，並在 timeout 設定較長。
多服務聚合（Dashboard）	同時顯示天氣、新聞、股價等資訊。	使用 asyncio.gather 並行呼叫多個 API，搭配 限流 防止雪崩。
資料同步（ETL）	定時抓取第三方資料寫入本地資料庫。	在 背景任務（BackgroundTasks）或 Celery 裡使用 httpx，確保任務獨立於 API 請求。

---

總結

FastAPI 與外部服務的整合不僅是 技術實作，更關係到 系統效能、可靠性與可維護性。透過本篇文章，我們學會：

• 為何必須使用 非同步 HTTP 客戶端（httpx、aiohttp）避免阻塞。
• 如何 建立共享 client、設定逾時、處理例外，以及 使用依賴注入 讓程式碼更乾淨。
• 利用 asyncio.gather、tenacity 等工具實現 並行呼叫與自動重試。
• 常見的陷阱（阻塞、資源洩漏、逾時缺失）以及對應的 最佳實踐。
• 具體的 實務場景，從支付、天氣到機器學習服務，都可以套用相同的模式。

只要遵循 「非同步、共享資源、統一錯誤與監控」 的四大原則，你的 FastAPI 應用就能在呼叫外部 API 時保持 高速、穩定且易於維護。祝你在開發旅程中玩得開心，寫出更好的服務！ 🚀