FastAPI

效能與最佳化（Performance & Optimization）

主題：快取（Redis / lru_cache）

---

簡介

在 Web API 開發中，效能往往是使用者體驗的關鍵。即使使用 FastAPI 這樣的高效框架，如果每一次請求都必須重新查詢資料庫、計算大量資料，仍然會產生不必要的延遲。
快取（Caching）正是解決這類瓶頸的常見手段：將「熱」的資料暫時保留在較快的存取層（記憶體或分散式快取服務），讓後續請求能直接命中快取，省去昂貴的 I/O 或運算成本。

本篇文章將從 Python 標準庫的 functools.lru_cache、分散式快取 Redis 兩個角度，說明在 FastAPI 中如何實作快取、避免常見陷阱，並提供可直接套用的範例程式碼，讓讀者能在自己的專案裡快速提升效能。

---

核心概念

1. 為什麼要快取？

需求	若不快取的結果	快取帶來的好處
頻繁讀取的參考資料（如商品分類、字典表）	每次都查 DB → 讀寫壓力、延遲提升	直接從記憶體或 Redis 讀取，毫秒級回應
計算成本高的報表或統計	每次請求都重新計算 → CPU 高負載	計算一次後存入快取，後續直接回傳
需要在多個服務間共享資料	各服務自行計算 → 資料不一致	Redis 作為中心化快取，保證一致性

重點：快取不是「永遠」快，而是「在適當的時機」快。快取策略（何時寫入、何時失效）必須根據業務需求設計。

---

2. functools.lru_cache：最簡單的本機快取

lru_cache（Least Recently Used）是一個裝飾器，會把函式的輸入與返回值保存在記憶體中，當快取已滿時會淘汰最久未使用的項目。它非常適合 純函式（不涉及 I/O）或 同步 的計算密集型工作。

範例 1：同步端點使用 lru_cache

# main.py
from fastapi import FastAPI
from functools import lru_cache
import time

app = FastAPI()

# 假設這是一個計算成本較高的函式
@lru_cache(maxsize=128)          # 最多快取 128 筆不同的參數組合
def heavy_computation(x: int) -> int:
    time.sleep(2)                # 模擬耗時的運算
    return x * x

@app.get("/square/{value}")
def get_square(value: int):
    """
    取得 value 的平方，第一次會慢 2 秒，之後會快取結果
    """
    result = heavy_computation(value)
    return {"value": value, "square": result}

說明

• @lru_cache 只會在 相同參數 時命中快取。
• maxsize 設為 None 時會無限制快取，但會佔用越來越多記憶體，需謹慎使用。

---

3. 非同步環境下的快取：async_lru

FastAPI 支援非同步路由，若在非同步函式中直接使用同步的 lru_cache 會造成 阻塞。此時可以使用第三方套件 async-lru（或 aiocache）提供的非同步快取裝飾器。

範例 2：非同步端點使用 async_lru

# async_main.py
from fastapi import FastAPI
from async_lru import alru_cache
import asyncio

app = FastAPI()

@alru_cache(maxsize=64)
async def async_heavy_computation(x: int) -> int:
    await asyncio.sleep(2)   # 非同步的延遲，模擬 I/O 或計算
    return x + 10

@app.get("/add10/{num}")
async def add_ten(num: int):
    """
    非同步版的「加十」運算，第一次會延遲 2 秒，之後快取命中
    """
    result = await async_heavy_computation(num)
    return {"input": num, "result": result}

提示：async_lru 內部仍使用字典保存快取資料，適合 單機 或 單進程 的情境；若部署多個 FastAPI 工作執行緒/容器，快取不會共享。

---

4. Redis：分散式、跨實例的快取

當服務需要水平擴展（多個容器或多台機器）時，單機記憶體快取已不夠。Redis 是最常見的 鍵值資料庫，提供高速的讀寫、過期機制、持久化選項，且支援 Pub/Sub、Lua 腳本 等進階功能。

4.1 建立 Redis 連線（使用 aioredis）

# redis_client.py
import aioredis
from fastapi import Depends

REDIS_URL = "redis://localhost:6379"

async def get_redis() -> aioredis.Redis:
    """
    FastAPI 的依賴項，確保每個請求都能取得同一個連線池
    """
    redis = await aioredis.from_url(REDIS_URL, decode_responses=True)
    try:
        yield redis
    finally:
        await redis.close()

說明

• decode_responses=True 讓取得的值自動轉成 str，省去手動解碼。
• 使用 依賴注入（Dependency Injection）可以讓路由函式直接取得 redis 實例，保持程式碼乾淨。

4.2 快取資料庫查詢結果

假設有一個商品資訊表，我們希望將常查詢的商品資料快取至 Redis，並設定 10 分鐘 的過期時間。

# product_api.py
from fastapi import APIRouter, Depends, HTTPException
from redis_client import get_redis
import json
import asyncpg  # PostgreSQL 非同步驅動

router = APIRouter()
DB_DSN = "postgresql://user:pass@localhost:5432/shop"

async def fetch_product_from_db(product_id: int) -> dict:
    conn = await asyncpg.connect(DB_DSN)
    row = await conn.fetchrow("SELECT id, name, price FROM products WHERE id=$1", product_id)
    await conn.close()
    if row:
        return dict(row)
    raise HTTPException(status_code=404, detail="Product not found")

@router.get("/product/{product_id}")
async def get_product(product_id: int, redis=Depends(get_redis)):
    cache_key = f"product:{product_id}"
    # 1️⃣ 嘗試從 Redis 讀取
    cached = await redis.get(cache_key)
    if cached:
        # 命中快取，直接回傳
        return json.loads(cached)

    # 2️⃣ 快取未命中 → 從 DB 讀取
    product = await fetch_product_from_db(product_id)

    # 3️⃣ 寫入 Redis，設定 600 秒過期
    await redis.set(cache_key, json.dumps(product), ex=600)

    return product

重點

• Cache‑Aside Pattern（先讀快取、未命中再查 DB，最後寫回快取）是最常見且易於維護的策略。
• ex=600 表示快取 600 秒後自動過期，避免過時資料長時間佔用快取空間。

---

5. 使用專門的 FastAPI 快取套件：fastapi-cache2

若想把快取邏輯抽象化、統一管理過期時間與序列化，可使用 fastapi-cache2（支援 Redis、Memory、Memcached）。以下示範如何在全局層面設定快取。

# app_with_fastapi_cache.py
from fastapi import FastAPI, Depends
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
import aioredis

app = FastAPI()

@app.on_event("startup")
async def startup():
    redis = await aioredis.from_url("redis://localhost:6379", decode_responses=True)
    FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")

@app.get("/weather/{city}")
@cache(expire=300)   # 快取 5 分鐘
async def get_weather(city: str):
    """
    假設此函式會呼叫外部天氣 API，耗時較長
    """
    # 這裡僅模擬
    import random, asyncio
    await asyncio.sleep(1)
    return {"city": city, "temp": random.randint(15, 30)}

優點

• 只需要在路由上加上 @cache 裝飾器，即可自動完成 Key 產生、序列化、過期管理。
• 支援 依賴注入，讓不同的端點可以使用不同的快取後端（Memory vs Redis）。

---

常見陷阱與最佳實踐

陷阱	說明	解決方案
快取過大導致記憶體不足	lru_cache(maxsize=None) 會持續累積	設定合理的 maxsize，或使用 TTL（Time‑to‑Live）機制
快取與資料不一致	資料表更新後快取仍保留舊值	在寫入或刪除資料時主動 刪除對應快取鍵（redis.delete(key)）
序列化錯誤	直接把 Python 物件塞入 Redis 會拋錯	使用 json.dumps / orjson 或 pickle（注意安全性）
阻塞非同步程式	在 async 路由裡使用同步的 time.sleep、requests 等	改用 await asyncio.sleep、httpx.AsyncClient
快取鍵衝突	不同資料使用相同鍵名導致覆寫	為每類型資料加上 前綴（如 product:{id}）
過期時間設置不當	設太長會保留過時資料、太短則失去快取效益	依據資料變動頻率與業務需求，使用 動態 TTL 或 滑動過期（Sliding expiration）

最佳實踐清單

1. 先評估快取收益：使用 A/B 測試或簡易的計時 (time.perf_counter) 確認快取前後延遲差距。
2. 統一快取鍵命名規則：例如 "{namespace}:{resource}:{id}"，方便日後清除與管理。
3. 結合失效通知：Redis 支援 Keyspace Notifications，可在資料變更時自動清除相關快取。
4. 監控快取命中率：利用 INFO stats 或 redis-cli 的 keyspace_hits/keyspace_misses 觀察效能。
5. 分層快取：先使用 本機 lru_cache 處理極高頻率的簡單計算，再把較大或跨實例的資料放入 Redis。

---

實際應用場景

場景	為何需要快取	建議實作方式
商品列表分頁	同一頁面的資料頻繁被瀏覽	使用 Redis 快取分頁結果，TTL 5 分鐘；更新商品時刪除相關快取鍵。
使用者權限驗證	每次請求都要檢查 JWT 與 DB 中的角色	把使用者 ID → 角色列表快取到 lru_cache（單機）或 Redis（多實例），TTL 10 分鐘。
外部 API 數據（如天氣、匯率）	第三方服務有速率限制或回應慢	使用 fastapi-cache2 搭配 Redis，設定 1 小時 TTL，減少外部呼叫次數。
報表統計（每日銷售額）	大量聚合計算，CPU 佔用高	先把統計結果寫入 Redis，使用 expire 或每日凌晨自動重算。
機器學習模型預測	模型推論需要載入大型模型檔案	把模型物件放入 lru_cache，確保同一個工作執行緒只載入一次。

---

總結

快取是提升 FastAPI 應用效能的 關鍵武器，從最簡單的 functools.lru_cache 到功能完整的 Redis，都各有適用情境：

• lru_cache → 單機、同步/非同步、計算密集型、資料不會跨實例共享。
• Redis → 分散式、跨容器、需要過期與持久化、支援 Pub/Sub 或 Lua。
• fastapi-cache2 → 想要快速在路由上套用快取、統一管理快取後端與 TTL。

在實作時，務必要注意 快取一致性、記憶體使用、鍵名規則與監控，避免因快取帶來的副作用抵消效能提升。透過上述概念與範例，你已具備在 FastAPI 專案中導入快取的完整知識，接下來就可以根據自己的業務需求，選擇最適合的快取策略，讓 API 的回應速度與可擴展性同步提升。祝開發順利！