FastAPI – 非同步程式設計（Async Programming）

主題：async context manager

---

簡介

在 FastAPI 中，非同步（async）已成為處理 I/O 密集型工作（例如資料庫、檔案、外部 API）時的標準做法。當我們需要在請求的生命週期內取得、使用、最後釋放資源時，async context manager（非同步上下文管理器）是最乾淨、最安全的工具。它讓資源的開啟與關閉自動配對，避免遺漏 await 或忘記釋放，從而減少記憶體洩漏與併發錯誤。

本篇文章將從概念說明、實作方式、常見陷阱與最佳實踐，一直到實務應用場景，完整介紹在 FastAPI 中如何運用 async context manager，讓你的 API 更具可讀性與可靠性。

---

核心概念

1. 為什麼需要 async context manager？

• 同步的 with 只能搭配實作 __enter__ / __exit__ 的類別，無法直接 await 內部的非同步操作。
• 非同步環境（如 async def）中，開啟資料庫連線、取得 Redis 客戶端、讀寫大型檔案等，都需要使用 await，因此必須使用 async with。
• async with 會在 __aenter__ 完成後才進入程式區塊，離開時自動呼叫 __aexit__，即使程式拋出例外也會保證資源被正確釋放。

2. 基本語法

async with my_async_context_manager() as resource:
    # 在此使用 resource，所有非同步操作都可以直接 await
    await resource.do_something()
# 離開區塊後，__aexit__ 會自動被呼叫

3. 實作方式

3.1 手寫 __aenter__ / __aexit__

class AsyncFile:
    """非同步檔案讀寫範例，使用 aiofiles 套件"""
    def __init__(self, path: str, mode: str = "r"):
        self.path = path
        self.mode = mode
        self._file = None

    async def __aenter__(self):
        import aiofiles
        self._file = await aiofiles.open(self.path, mode=self.mode)
        return self._file  # 回傳 aiofiles 的檔案物件

    async def __aexit__(self, exc_type, exc, tb):
        await self._file.close()
        # 若需要自行處理例外，可在此回傳 True 抑制
        return False

# 使用方式
async with AsyncFile("data.txt", "w") as f:
    await f.write("Hello, FastAPI async context manager!\n")

3.2 contextlib.asynccontextmanager 裝飾器

from contextlib import asynccontextmanager
import asyncpg

@asynccontextmanager
async def get_db_connection():
    """取得 PostgreSQL 連線，使用 asyncpg"""
    conn = await asyncpg.connect(dsn="postgresql://user:pw@localhost/db")
    try:
        yield conn
    finally:
        await conn.close()

# 在路由中使用
@app.get("/users/{uid}")
async def read_user(uid: int):
    async with get_db_connection() as conn:
        row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", uid)
        return dict(row) if row else {"error": "User not found"}

3.3 FastAPI 依賴（Dependency）結合 async context manager

FastAPI 允許把 async with 包裝成 依賴，讓每一次請求自動取得並釋放資源。

from fastapi import Depends, FastAPI

app = FastAPI()

async def db_session():
    async with get_db_connection() as conn:
        yield conn   # 交給路由使用，離開時自動關閉

@app.get("/items/")
async def list_items(conn=Depends(db_session)):
    rows = await conn.fetch("SELECT * FROM items")
    return [dict(r) for r in rows]

3.4 與 Redis（aioredis）結合

import aioredis
from contextlib import asynccontextmanager

@asynccontextmanager
async def redis_client():
    client = await aioredis.from_url("redis://localhost")
    try:
        yield client
    finally:
        await client.close()

@app.post("/cache/{key}")
async def set_cache(key: str, value: str, r=Depends(redis_client)):
    await r.set(key, value, ex=3600)  # 設定 1 小時過期
    return {"msg": "cached"}

3.5 同時管理多個資源

有時候一個請求需要同時取得 DB、Redis、外部 API 的 client。可以把多個 async with 串接，或使用自訂的「組合管理器」。

@asynccontextmanager
async def resources():
    async with get_db_connection() as db, redis_client() as redis:
        yield {"db": db, "redis": redis}

@app.get("/combo")
async def combo(res=Depends(resources)):
    db = res["db"]
    redis = res["redis"]
    # 兩個資源皆已正確開啟，使用完畢後會自動關閉
    await redis.set("last_access", "now")
    rows = await db.fetch("SELECT count(*) FROM logs")
    return {"logs": rows[0]["count"]}

---

常見陷阱與最佳實踐

陷阱	說明	解決方式
忘記 await __aenter__	直接使用 async with MyCM() 而忘記 await 會得到 coroutine 而非資源。	確保 async with 前已在 async def 中，且 MyCM 實作正確的 __aenter__。
在同步函式中使用 async with	會拋出 RuntimeError: cannot use 'await' outside of async function。	把需要非同步資源的程式碼搬到 async def，或使用 anyio.run 包裝。
例外未被正確傳遞	__aexit__ 回傳 True 會抑制例外，可能隱藏錯誤。	大多數情況下 返回 False（或不回傳），只在確定要捕捉並處理的例外時才返回 True。
資源重用導致競爭	同一個 async context manager 物件被多個請求同時 await，可能產生 race condition。	每一次請求都 建立新的實例（如使用依賴工廠），或在管理器內部使用鎖 (asyncio.Lock)。
忘記關閉外部資源	若 __aexit__ 中忘記 await client.close()，連線池會持續增長。	統一在 finally 區塊 中關閉，並在測試環境確認 close 被呼叫。

最佳實踐：

1. 盡量使用 contextlib.asynccontextmanager：語法簡潔、易於維護。
2. 把資源取得封裝成 FastAPI 依賴，讓每一次請求自動取得、釋放。
3. 在 __aexit__ 中使用 try/except 捕捉關閉過程的例外，避免因關閉失敗而導致資源泄漏。
4. 加入型別提示（-> AsyncGenerator[Client, None]），提升 IDE 輔助與文件可讀性。
5. 在測試環境使用 pytest-asyncio，驗證 async with 的正確行為。

---

實際應用場景

場景	為何使用 async context manager
資料庫交易（transaction）	需要在同一個連線中 BEGIN → … → COMMIT/ROLLBACK，async with 能保證在例外時自動 ROLLBACK。
檔案上傳與處理	讀取大型檔案時使用 aiofiles，async with 確保檔案句柄即使在處理過程拋錯也會被關閉。
外部 API 客戶端	像 httpx.AsyncClient 需要在使用完畢後關閉連線池，async with httpx.AsyncClient() as client: 為標準寫法。
多資源協調	同時使用 DB、Redis、Message Queue 時，透過自訂的組合管理器一次取得全部，保持程式碼簡潔。
測試用的臨時資源	在測試套件中使用 async with 建立臨時資料表或測試資料庫，測試結束自動清理。

---

總結

• async context manager 為 FastAPI 的非同步程式設計提供了「安全、可讀、可擴」的資源管理方式。
• 透過 __aenter__ / __aexit__ 或 contextlib.asynccontextmanager，我們可以把 取得 → 使用 → 釋放 的流程完整封裝，讓每一次請求只關注業務邏輯。
• 在實務開發中，將資源取得寫成依賴（Dependency） 是最常見且最推薦的做法，能確保資源的生命週期與請求緊密相連。
• 注意常見的陷阱（忘記 await、在同步函式中使用、例外抑制等），遵循最佳實踐，就能寫出既高效又穩定的 FastAPI 服務。

希望本篇文章能幫助你在 FastAPI 專案中熟練運用 async context manager，讓 API 更具可維護性與效能。祝開發順利！ 🚀