FastAPI 依賴注入系統：`yield` Dependencies（用於建立 / 清理資源）

簡介

在 Web 框架中，依賴注入（Dependency Injection, DI） 是讓程式碼保持乾淨、可測試、易維護的核心機制。FastAPI 不僅提供了簡潔的 DI 寫法，還支援 yield 依賴，讓開發者可以在同一個函式裡完成 資源的建立 與 資源的釋放（清理）工作。

為什麼 yield 依賴如此重要？

統一管理資源生命週期：資料庫連線、檔案句柄、Redis 客戶端等，都需要在請求結束時關閉或回收。使用 yield，FastAPI 會自動在回傳結果後執行清理程式碼，避免忘記釋放資源而產生記憶體泄漏或連線耗盡的問題。
簡化測試：測試時只要呼叫相同的依賴函式，就能得到已建立好的資源，同時保證測試結束後自動清理，保持測試環境的潔淨。
提升效能：透過依賴的快取（Depends 的 cache 參數），同一個請求內的多個路由可以共用同一個已建立的資源，減少重複建立的開銷。

以下內容會一步步說明 yield 依賴的原理、寫法以及實務應用，讓你在 FastAPI 專案中能夠安全、有效地管理各種外部資源。

核心概念

1. `yield` 依賴的運作機制

在 FastAPI 中，依賴函式可以是 同步函式、非同步函式，或 產生器函式（使用 yield）。當 FastAPI 呼叫產生器函式時：

進入階段：先執行 yield 前的程式碼，建立資源（例如 engine.connect()）。
回傳階段：yield 後的值會被注入到需要此依賴的路由或其他依賴中。
退出階段：當請求結束（成功或拋出例外）時，FastAPI 會繼續執行產生器函式中 yield 後的程式碼，用於清理資源（例如 connection.close()）。

重點：yield 後的程式碼 一定會被執行，即使路由拋出例外，這保證了資源一定會被釋放。

2. 同步 vs 非同步 `yield` 依賴

同步 yield：使用普通的 def，適合傳統的阻塞式資源（如 SQLite 的同步連線）。
非同步 yield：使用 async def，適合 async 驅動的資源（如 asyncpg、aioredis）。
選擇原則：若資源本身支援 async，請使用 async def；若資源是阻塞的，使用同步函式即可。

3. 快取 (`cache`) 與作用域 (`scope`)

FastAPI 的依賴預設會快取（在同一次請求內只建立一次），但你可以透過 Depends(..., use_cache=False) 取消快取，或在 Depends 裡設定 scope="session"、scope="app" 等，以控制依賴的生命週期。

程式碼範例

以下示範 5 個常見且實用的 yield 依賴範例，涵蓋同步、非同步、資料庫、檔案、以及第三方服務。

範例 1️⃣ 同步資料庫連線（SQLite）

# db.py
import sqlite3
from fastapi import Depends

def get_db():
    """同步建立 SQLite 連線，請求結束後自動關閉。"""
    conn = sqlite3.connect("example.db")
    try:
        yield conn
    finally:
        conn.close()

# main.py
from fastapi import FastAPI, Depends
from db import get_db

app = FastAPI()

@app.get("/users")
def read_users(db: sqlite3.Connection = Depends(get_db)):
    cursor = db.cursor()
    cursor.execute("SELECT id, name FROM users")
    return cursor.fetchall()

說明：yield 前建立連線，finally 區塊在請求結束時關閉連線，確保不會遺漏。

範例 2️⃣ 非同步 PostgreSQL 連線（`asyncpg`）

# async_db.py
import asyncpg
from fastapi import Depends

async def get_async_connection():
    """非同步取得 PostgreSQL 連線，使用 async with 確保釋放。"""
    conn = await asyncpg.connect(user="postgres", password="secret",
                                 database="testdb", host="127.0.0.1")
    try:
        yield conn
    finally:
        await conn.close()

# main_async.py
from fastapi import FastAPI, Depends
from async_db import get_async_connection

app = FastAPI()

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

說明：非同步資源必須使用 await 釋放，FastAPI 會在 await conn.close() 前等候完成。

範例 3️⃣ 檔案處理（上傳暫存檔）

# file_dep.py
import tempfile
from fastapi import Depends, UploadFile

def get_temp_file():
    """建立臨時檔案，請求結束後自動刪除。"""
    tmp = tempfile.NamedTemporaryFile(delete=False)
    try:
        yield tmp.name
    finally:
        tmp.close()
        os.remove(tmp.name)

# main_file.py
from fastapi import FastAPI, Depends, File, UploadFile
from file_dep import get_temp_file

app = FastAPI()

@app.post("/upload")
async def upload_file(file: UploadFile = File(...),
                      temp_path: str = Depends(get_temp_file)):
    # 把上傳的內容寫入暫存檔
    with open(temp_path, "wb") as buffer:
        data = await file.read()
        buffer.write(data)
    # 此時 temp_path 已經寫好，接下來可以做其他處理
    return {"filename": file.filename, "temp_path": temp_path}

說明：即使上傳過程中拋出例外，finally 區塊仍會刪除暫存檔，避免磁碟被佔滿。

範例 4️⃣ Redis 客戶端（`aioredis`）與自訂快取

# redis_dep.py
import aioredis
from fastapi import Depends

async def get_redis():
    """取得 Redis 連線，使用 app scope 只建立一次。"""
    redis = await aioredis.from_url("redis://localhost")
    try:
        yield redis
    finally:
        await redis.close()

# main_redis.py
from fastapi import FastAPI, Depends, HTTPException
from redis_dep import get_redis

app = FastAPI()

@app.get("/cache/{key}")
async def read_cache(key: str, redis = Depends(get_redis)):
    value = await redis.get(key)
    if value is None:
        raise HTTPException(status_code=404, detail="Key not found")
    return {"key": key, "value": value.decode()}

說明：將 get_redis 設為 app 範圍（預設 request），可在整個應用程式生命週期內共享同一個連線池，減少連線開銷。

範例 5️⃣ 交易（Transaction）管理（SQLAlchemy 2.x Async）

# transaction.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from fastapi import Depends

DATABASE_URL = "postgresql+asyncpg://user:pwd@localhost/db"

engine = create_async_engine(DATABASE_URL, echo=False)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db_session():
    """取得資料庫 Session，並在請求結束時自動 rollback/commit。"""
    async with AsyncSessionLocal() as session:
        async with session.begin():
            # 這裡的 yield 會把 session 注入給路由
            yield session
        # session 會在離開 context 時自動 commit

# main_tx.py
from fastapi import FastAPI, Depends, HTTPException
from transaction import get_db_session
from models import User  # 假設已定義 ORM 模型

app = FastAPI()

@app.post("/users")
async def create_user(name: str, db: AsyncSession = Depends(get_db_session)):
    new_user = User(name=name)
    db.add(new_user)
    # commit 會在 context 結束時自動執行
    return {"id": new_user.id, "name": new_user.name}

說明：使用 async with session.begin() 可以保證交易在成功時自動 commit、失敗時自動 rollback，不需要手動寫 try/except。

常見陷阱與最佳實踐

陷阱	說明	建議的做法
忘記 `finally` / `await`	產生器結束前未正確釋放資源，導致連線泄漏。	一定在 `yield` 後使用 `finally`（同步）或 `await`（非同步）來關閉資源。
在 `yield` 前拋出例外	若建立資源時失敗，FastAPI 不會執行 `finally`，但會直接傳遞例外。	在建立階段加入例外捕獲，並回傳適當的 HTTPException。
快取導致共享狀態	依賴被快取後，跨請求共享同一個物件，可能產生競爭條件。	只在無狀態或線程安全的資源上使用快取；對於有狀態的連線（如 DB session）保留預設 `request` 範圍。
使用阻塞 I/O 在 async 依賴	在 `async def` 內呼叫同步阻塞函式，會阻塞事件迴圈。	分離：同步依賴使用 `def`，非同步依賴使用 `async def`，或使用 `run_in_threadpool` 包裝阻塞呼叫。
過度依賴 `yield`	把所有程式碼都寫在同一個產生器裡，導致可讀性下降。	把資源建立與資源清理分離成兩個小函式，然後在產生器中呼叫，保持單一職責。

最佳實踐清單

明確標註資源類型：同步使用 def、非同步使用 async def。
使用 try/finally（或 async with）保證清理程式碼一定執行。
僅在需要時啟用快取：對於資料庫 Session、交易等，保留預設的 request 快取。
將資源建立抽離：例如把 engine = create_engine(...) 放在模組層級，只在產生器裡取得連線。
測試資源釋放：在單元測試中使用 TestClient，確認每次請求後資源已被關閉（可以透過 mock 或 log 觀察）。

實際應用場景

場景	為何適合使用 `yield` 依賴	範例簡述
WebSocket 連線	每個連線需要長時間保持資料庫或 Redis 連線，且在斷線時必須釋放。	建立 `async def get_ws_redis()`，在 `yield` 前取得連線，`finally` 中關閉。
背景任務（BackgroundTasks）	任務完成後需要清理臨時檔案或釋放雲端儲存的資源。	`def get_temp_dir()` 產生臨時目錄，`finally` 刪除目錄。
多租戶系統	每個請求依租戶切換資料庫 schema，結束時必須回復或關閉連線。	`def get_tenant_db(tenant_id: str = Depends(get_current_tenant))` 使用 `yield` 切換 schema，最後恢復預設。
外部 API 客戶端	呼叫第三方服務前需要初始化 client，完成後需要關閉連線池。	`async def get_http_client()` 產生 `httpx.AsyncClient()`，`finally` `await client.aclose()`。
定期清理任務	使用 FastAPI 的 `@repeat_every` 時，需要在每次執行前取得 DB 連線，執行完後釋放。	`def get_db()` 產生 `SessionLocal()`，在定時任務裡 `with Depends(get_db) as db:` 使用。

總結

yield 依賴是 FastAPI 中最強大的資源管理工具，它把「建立」與「清理」的程式碼綁在同一個函式裡，保證在任何情況下（成功、失敗、例外）都能正確釋放資源。
同步與非同步的寫法只差 def / async def，但 務必使用 try/finally 或 async with 來確保清理程式碼一定被執行。
快取與作用域的設定讓你可以根據資源的特性選擇 一次請求內共享 或 全域單例，避免不必要的重複建立。
常見的陷阱（忘記釋放、錯誤的快取、阻塞 I/O）只要遵守最佳實踐，就能輕鬆避免。

掌握了 yield 依賴後，你就可以在 FastAPI 中自信地處理資料庫、Redis、檔案、WebSocket、外部 API 等各種外部資源，寫出 高效、可靠且易於維護 的服務。快把這些範例搬到自己的專案中實作，體驗依賴注入帶來的開發快感吧！ 🚀

FastAPI 依賴注入系統：yield Dependencies（用於建立 / 清理資源）

簡介