FastAPI 與外部服務整合

主題：Redis 快取

---

簡介

在現代 Web API 中，效能與可擴充性往往是使用者體驗的關鍵。即使是寫得很好的 FastAPI 應用，若每一次請求都必須直接查詢資料庫或遠端服務，仍可能因 I/O 延遲而造成回應緩慢。這時，快取 (Cache) 成為提升效能的第一道防線，而 Redis 以其高速、支援多種資料結構、原生支援 TTL（Time‑To‑Live）等特性，成為最常見的快取解決方案。

本篇文章將從 概念、實作、最佳實踐 逐步說明，如何在 FastAPI 專案中無縫整合 Redis，讓讀者能夠在真實專案裡快速部署、測試與優化快取機制。文章適合 初學者 了解基本流程，也提供 中級開發者 進一步調校與擴充的範例。

---

核心概念

1. 為什麼需要快取

• 降低資料庫壓力：相同的查詢只會在快取命中時直接回傳，減少 DB 讀寫次數。
• 縮短回應時間：Redis 把資料存放在記憶體中，讀寫延遲通常在 毫秒以下，遠快於磁碟 I/O。
• 提升可用性：即使資料庫暫時不可用，已快取的資料仍能提供服務，提升系統韌性。

小技巧：先針對「熱點資料」(頻繁被讀取且變化不大的資料) 建立快取，逐步擴展至其他 API。

2. Redis 基礎

Redis（Remote Dictionary Server）是一個 鍵值 (Key‑Value) 資料庫，支援字串、哈希、列表、集合、有序集合等多種資料結構。常見的指令包括：

指令	說明
SET key value [EX seconds]	設定字串，EX 為過期時間
GET key	取得字串值
HSET hash field value	在哈希中設定欄位
LPUSH list value	在列表左側推入元素
EXPIRE key seconds	為鍵設定 TTL

在 Python 中，我們通常使用 redis-py（同步）或 aioredis（非同步）兩個套件與 Redis 溝通。

3. 在 FastAPI 中使用 Redis

3.1 安裝與連線

pip install fastapi uvicorn redis[async]   # 同步與非同步套件皆安裝

# redis_client.py
import os
import redis
import aioredis
from functools import lru_cache

@lru_cache()
def get_sync_redis() -> redis.Redis:
    """
    建立同步 Redis 連線，使用 LRU cache 只會在程式啟動時建立一次
    """
    return redis.Redis(
        host=os.getenv("REDIS_HOST", "localhost"),
        port=int(os.getenv("REDIS_PORT", 6379)),
        db=0,
        decode_responses=True,   # 直接回傳 str 而非 bytes
    )

async def get_async_redis() -> aioredis.Redis:
    """
    建立非同步 Redis 連線，使用單例模式避免重複連線
    """
    return await aioredis.from_url(
        f"redis://{os.getenv('REDIS_HOST','localhost')}:{os.getenv('REDIS_PORT','6379')}/0",
        encoding="utf-8",
        decode_responses=True,
    )

註：在正式環境建議使用 環境變數 或 Docker secret 管理 Redis 密碼與主機資訊。

3.2 範例一：簡易快取裝飾器

# cache_decorator.py
import json
import hashlib
from typing import Callable, Any
from fastapi import Depends
from redis_client import get_sync_redis

def cache(ttl: int = 60):
    """
    以裝飾器方式為任意函式加入快取功能
    - `ttl` 為快取存活秒數，預設 60 秒
    """
    def decorator(func: Callable):
        def wrapper(*args, **kwargs) -> Any:
            # 產生唯一的快取鍵，使用函式名稱 + 參數雜湊
            raw_key = f"{func.__name__}:{json.dumps([args, kwargs], sort_keys=True)}"
            cache_key = hashlib.sha256(raw_key.encode()).hexdigest()

            r = get_sync_redis()
            cached = r.get(cache_key)
            if cached is not None:
                # 命中快取，直接回傳解碼後的結果
                return json.loads(cached)

            # 未命中，執行原函式
            result = func(*args, **kwargs)

            # 將結果寫入快取，並設定 TTL
            r.setex(cache_key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

# main.py
from fastapi import FastAPI
from cache_decorator import cache

app = FastAPI()

@cache(ttl=120)   # 兩分鐘快取
def heavy_computation(x: int, y: int) -> dict:
    # 假設這是一段耗時的計算或外部 API 呼叫
    import time; time.sleep(2)
    return {"result": x + y, "timestamp": time.time()}

@app.get("/add")
def add(a: int, b: int):
    """
    透過裝飾器快取計算結果，第二次相同參數只會即時回傳
    """
    return heavy_computation(a, b)

重點：

• 使用 hashlib.sha256 產生唯一且固定長度的快取鍵，避免鍵過長導致 Redis 效能下降。
• json.dumps / json.loads 讓快取內容可序列化，適用於大多數 Python 資料型別。

3.3 範例二：依賴注入 (Depends) 結合 Redis

FastAPI 的 依賴注入 機制讓我們可以在路由函式中直接取得 Redis 連線，保持程式碼乾淨且易於測試。

# deps.py
from fastapi import Depends
from redis_client import get_sync_redis, get_async_redis

def redis_sync_dep():
    """同步 Redis 依賴，回傳 Redis 實例"""
    return get_sync_redis()

async def redis_async_dep():
    """非同步 Redis 依賴，回傳 aioredis 實例"""
    return await get_async_redis()

# main.py (續)
from fastapi import Depends
from deps import redis_sync_dep, redis_async_dep
import json

@app.get("/user/{user_id}")
def get_user(user_id: int, r: redis.Redis = Depends(redis_sync_dep)):
    """
    先從快取取得使用者資料，若未命中再查 DB（此處以假資料代替）
    """
    cache_key = f"user:{user_id}"
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    # 假設從資料庫撈資料
    user_data = {"id": user_id, "name": f"User{user_id}", "age": 30}
    r.setex(cache_key, 300, json.dumps(user_data))  # 5 分鐘快取
    return user_data

說明：

• Depends 讓路由函式不需要自行呼叫 get_sync_redis()，測試時只要替換依賴即可。
• 透過 setex 同時設定 值 與 過期時間，避免忘記呼叫 expire。

3.4 範例三：TTL 與過期策略

不同的資料有不同的「新鮮度」需求，Redis 提供兩種主要的過期策略：

策略	說明
TTL (Time‑To‑Live)	固定秒數過期，最常用於暫存計算結果。
LRU (Least‑Recently‑Used)	Redis 本身的記憶體淘汰機制，適合大量短暫快取。

以下示範如何在 Redis 設定 中啟用 LRU，並在程式碼層面使用 TTL。

# redis.conf (或 Docker 環境變數)
maxmemory 256mb          # 限制 Redis 使用的最大記憶體
maxmemory-policy allkeys-lru   # 以 LRU 淘汰所有鍵

# main.py (續)
@app.get("/stats")
def get_stats(r: redis.Redis = Depends(redis_sync_dep)):
    """
    取得最近 10 筆 API 呼叫次數的統計，使用 Redis 的有序集合 (ZSET)
    """
    # ZINCRBY 會自動為 key 增加分數，若 key 不存在則建立
    r.zincrby("api:call_counts", 1, "stats_endpoint")
    # 取得分數最高的前 10 名
    top10 = r.zrevrange("api:call_counts", 0, 9, withscores=True)
    return {"top10": top10}

• ZINCRBY 讓我們可以 即時累計 呼叫次數，且不需要額外的資料表。
• 透過 maxmemory-policy allkeys-lru，當記憶體滿時，Redis 會自動淘汰最久未使用的鍵，降低手動管理的負擔。

3.5 範例四：批次寫入與管線 (pipeline)

當一次需要寫入多筆資料時，直接呼叫多次 SET 會產生大量網路往返，管線 能把多個指令合併成一次請求，顯著提升效能。

# batch_cache.py
from redis_client import get_sync_redis

def cache_bulk_user_profiles(profiles: list[dict], ttl: int = 600):
    """
    批次快取多筆使用者資料，使用 pipeline 減少 RTT
    - `profiles` 為 [{'id': 1, 'name': 'Alice'}, ...] 結構
    """
    r = get_sync_redis()
    pipe = r.pipeline(transaction=False)   # transaction=False 可避免鎖定
    for p in profiles:
        key = f"user:{p['id']}"
        pipe.setex(key, ttl, json.dumps(p))
    pipe.execute()   # 一次性送出所有指令

實務建議：若資料量非常大（>10,000 筆），可分批 (例如每 500 筆一批) 送出 pipeline，避免單次指令過多造成記憶體壓力。

3.6 範例五：非同步使用 aioredis（適用於高併發）

FastAPI 天生支援 非同步，配合 aioredis 可以在同一事件迴圈內完成 I/O，提升每秒請求數。

# async_cache.py
import json
from fastapi import FastAPI, Depends
from deps import redis_async_dep
import aioredis

app = FastAPI()

@app.get("/async/user/{uid}")
async def async_get_user(uid: int, r: aioredis.Redis = Depends(redis_async_dep)):
    cache_key = f"user:{uid}"
    cached = await r.get(cache_key)
    if cached:
        return json.loads(cached)

    # 假設此處有非同步 DB 查詢
    user = {"id": uid, "name": f"AsyncUser{uid}"}
    await r.setex(cache_key, 300, json.dumps(user))
    return user

• await r.get(...)、await r.setex(...) 完全不會阻塞事件迴圈，適合 大量同時連線 的情境。
• 若在同一請求內需要多次 Redis 操作，可使用 async with r.pipeline() as pipe: 取得非同步管線。

---

常見陷阱與最佳實踐

陷阱	說明	解決方案
快取鍵衝突	不同功能使用相同鍵名，導致資料被覆寫。	為每個模組或功能加上 命名空間前綴（如 user:, product:）。
忘記設定 TTL	快取永遠不會過期，最終會佔滿記憶體。	預設 必須設定 EX 或 SETEX，或在 Redis 設定 maxmemory-policy。
序列化不一致	使用不同的編碼方式（pickle vs json）導致讀取錯誤。	統一使用 JSON（或 MessagePack）並在 Redis 端設定 decode_responses=True。
同步阻塞	在非同步路由中使用同步 redis-py 會阻塞事件迴圈。	依需求選擇 aioredis；若必須混用，使用 run_in_threadpool 包裝。
大量鍵產生	每個請求都產生唯一鍵（如 UUID）導致快取失效。	設計適當的快取策略：只對「可重複」的查詢快取，對一次性結果直接回傳。

最佳實踐

1. 命名規則：<domain>:<entity>:<identifier>（例如 order:detail:12345）。
2. 分層快取：先在 應用層（裝飾器）快取，必要時再在 資料庫層（ORM）加入快取。
3. 監控與指標：使用 INFO, MONITOR 或 Prometheus Redis Exporter 收集 hit/miss 比例，持續優化。
4. 安全性：在生產環境設定 密碼、ACL，避免未授權存取。
5. 測試：將 Redis 依賴抽象成介面，在單元測試中使用 fakeredis 或 mock，確保快取行為可預測。

---

實際應用場景

場景	為何需要 Redis 快取	具體實作
商品列表頁	商品資料變動頻率低，且訪問量高。	使用 SETEX 快取整頁 JSON，TTL 設為 5 分鐘。
使用者權限驗證	每次請求都要檢查權限，DB 查詢成本高。	把使用者 ID 與權限列表快取於 Redis，TTL 設為 30 分鐘或使用 Redis Hash。
第三方 API 整合	受限於外部服務的 rate‑limit。	把外部 API 回傳結果快取，失效前直接使用快取，避免重複呼叫。
即時排行榜	需要即時統計點擊或投票。	使用 Sorted Set (ZSET)，配合 ZINCRBY、ZREVRANGE 產生前 N 名。
會話 (Session) 管理	分散式服務需要共享會話狀態。	把 JWT 失效列表或 session ID 存於 Redis，利用 EXPIRE 自動失效。

---

總結

Redis 為 FastAPI 提供了一條高效、彈性的快取管道，能在 降低資料庫壓力、縮短回應時間、提升系統韌性 上發揮關鍵作用。本文從 安裝、連線、快取裝飾器、依賴注入、TTL 與淘汰策略、批次寫入、非同步操作 等角度提供完整範例，並列出常見陷阱與最佳實踐，協助開發者在實務專案中快速落地。

實踐建議：先在開發環境以 單一功能（如使用者資訊）驗證快取邏輯，再逐步擴展至全站快取；同時持續監控 hit/miss 數據，調整 TTL、命名空間與淘汰策略，讓你的 FastAPI 應用在高併發情境下依舊保持高速與穩定。

祝開發順利，快取無礙！