本文 AI 產出，尚未審核

FastAPI 請求生命週期 – Lifespan Events（startup / shutdown）

簡介

在 Web 應用程式的執行過程中，啟動（startup） 與 關閉（shutdown） 兩個階段往往被忽略，卻是資源管理、外部服務連線、快取初始化等關鍵工作必須完成的時機。FastAPI 提供了 lifespan 事件機制，讓開發者能在應用程式啟動時執行一次性的設定，並在服務關閉前釋放資源，確保系統在高可用環境下依然穩定運作。

本篇文章將深入說明 FastAPI 的 lifespan events，從概念、實作範例，到常見陷阱與最佳實踐，幫助初學者與中階開發者快速掌握這項功能，並在實務專案中有效運用。

核心概念

1. 為什麼需要 Lifespan Events？

  • 資源初始化：資料庫連線池、Redis 快取、訊息佇列（RabbitMQ、Kafka）等外部資源往往需要在應用程式第一次接收請求前建立。
  • 一次性設定：載入機器學習模型、讀取大型設定檔、產生加密金鑰等操作不宜在每個請求裡重複執行，會嚴重拖慢回應時間。
  • 資源釋放：服務關閉時要正確關閉連線、釋放記憶體、寫入結束日誌，避免資源洩漏或資料遺失。

FastAPI 的 lifespan 事件正是為了在 應用程式的生命週期 中提供掛鉤（hook），讓開發者在正確的時機點執行上述工作。

2. 兩種寫法：@app.on_event 與 lifespan Context Manager

寫法 語法 說明
@app.on_event("startup") / @app.on_event("shutdown") 裝飾函式 最常見的寫法，適合簡單的同步或非同步函式。
lifespan 參數（async with） FastAPI(lifespan=lifespan_func) 以 context manager 形式包住整個應用，支援更細緻的資源管理，尤其在測試或多階段啟動流程時更有彈性。

下面分別說明兩種寫法的使用方式與差異。

3. 使用 @app.on_event 的基本範例

# app_startup_shutdown.py
from fastapi import FastAPI
import asyncio

app = FastAPI()

# ---------- Startup ----------
@app.on_event("startup")
async def startup_event():
    """
    應用程式啟動時會執行此函式。
    這裡示範建立資料庫連線池與載入模型。
    """
    # 假設有一個 async 的資料庫連線建立函式
    app.state.db = await async_create_db_pool()
    # 載入機器學習模型（同步範例，使用 threadpool 包裝）
    loop = asyncio.get_event_loop()
    app.state.model = await loop.run_in_executor(None, load_heavy_model)
    print("🚀 Startup 完成：資料庫與模型已初始化")

# ---------- Shutdown ----------
@app.on_event("shutdown")
async def shutdown_event():
    """
    應用程式關閉前執行此函式。
    用於釋放資源、寫入日誌等。
    """
    await app.state.db.close()
    # 若模型佔用大量記憶體，可手動釋放
    del app.state.model
    print("🛑 Shutdown 完成：資源已釋放")

重點：

  • app.state 是 FastAPI 提供的共享屬性容器，適合存放跨請求共用的物件。
  • startup、shutdown 必須是 非同步（async def）或同步函式，FastAPI 會自動偵測。

4. 使用 lifespan Context Manager 的進階範例

# app_lifespan.py
from fastapi import FastAPI
from contextlib import asynccontextmanager
import asyncpg
import asyncio

@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    這個 async context manager 包住整個 FastAPI 應用的生命週期。
    進入 (yield 前) 為 startup，離開 (yield 後) 為 shutdown。
    """
    # ---------- Startup ----------
    print("🔧 正在建立 PostgreSQL 連線池…")
    app.state.pg_pool = await asyncpg.create_pool(dsn="postgresql://user:pw@localhost/db")
    print("✅ PostgreSQL 連線池已建立")

    # 載入外部設定檔（例如 YAML）
    import yaml, pathlib
    config_path = pathlib.Path(__file__).parent / "config.yaml"
    app.state.config = yaml.safe_load(config_path.read_text())
    print("📄 設定檔已載入")

    # 交給 FastAPI 處理請求
    yield

    # ---------- Shutdown ----------
    print("🔒 正在關閉 PostgreSQL 連線池…")
    await app.state.pg_pool.close()
    print("🗑️ PostgreSQL 連線池已關閉")
    # 若有其他資源（如 Redis）亦在此釋放

app = FastAPI(lifespan=lifespan)

說明：

  • lifespan 以 async with 的方式包住整個應用，yield 前的程式碼相當於 startup，yield 後則是 shutdown。
  • 這種寫法的好處是 一次性呈現完整生命週期流程，特別適合在測試環境中使用 TestClient 時自動執行初始化與清理。

5. 取得 Lifespan 資源的方式

在路由函式或依賴項（dependency）裡，我們常需要使用 startup 時建立的資源。以下示範如何透過 依賴注入 取得資料庫連線池：

# dependencies.py
from fastapi import Depends, FastAPI

def get_db_pool(app: FastAPI = Depends()):
    """
    從 app.state 取得已建立的資料庫連線池。
    若尚未初始化，會拋出 RuntimeError。
    """
    if not hasattr(app.state, "pg_pool"):
        raise RuntimeError("資料庫連線池尚未初始化")
    return app.state.pg_pool

# router.py
from fastapi import APIRouter, Depends
from dependencies import get_db_pool
import asyncpg

router = APIRouter()

@router.get("/users/{user_id}")
async def read_user(user_id: int, pool: asyncpg.Pool = Depends(get_db_pool)):
    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
        return dict(row) if row else {"error": "User not found"}

透過 Depends(get_db_pool)，FastAPI 會自動把 app 注入 get_db_pool，讓路由函式直接取得已初始化的資源。

常見陷阱與最佳實踐

陷阱 說明 建議的最佳實踐
忘記在 shutdown 釋放資源 服務關閉時仍保留連線或檔案句柄，導致資源洩漏。 在 shutdown 中 必須 呼叫 close()、await 釋放所有非同步資源。
在 startup 中使用阻塞 I/O 阻塞事件迴圈會導致服務無法接受請求。 使用 asyncio.to_thread、run_in_executor 或改寫為非同步 API。
依賴項在 startup 前使用 若路由在應用啟動前就被呼叫（例如單元測試未使用 TestClient），會找不到 app.state 中的資源。 在測試中使用 with TestClient(app) as client:，或在 dependency 中加入 容錯機制（如上例的 RuntimeError）。
多個 startup / shutdown 裝飾器執行順序不明 若有多個 @app.on_event("startup")，執行順序依註冊順序，但不易掌控。 統一管理：將相關資源的初始化與釋放封裝在同一個 lifespan context manager，或使用 module-level 初始化函式。
在 lifespan 中忘記 yield 沒有 yield 會讓 FastAPI 在啟動後直接結束，導致服務無法運行。 確保 lifespan 包含 yield，且 只 有一個 yield 點。

其他最佳實踐

  1. 使用 app.state 而非全域變數
    app.state 會隨應用實例而存在，避免在多個 FastAPI 實例（例如測試環境）中產生衝突。

  2. 將資源初始化抽離成獨立函式
    讓 startup 只負責呼叫，保持程式碼可讀性與可測試性。

  3. 加入日誌與監控
    在 startup、shutdown 中加入 structured logging（例如使用 loguru）以及 Prometheus metrics，方便觀測服務的啟動與關閉時間。

  4. 考慮容錯與重試
    外部服務（資料庫、Redis）如果在啟動時暫時不可用，應該實作 重試機制，避免服務因一次性失敗而無法啟動。

# 重試範例
import tenacity

@tenacity.retry(stop=tenacity.stop_after_attempt(5), wait=tenacity.wait_fixed(2))
async def create_pg_pool():
    return await asyncpg.create_pool(dsn="postgresql://user:pw@localhost/db")

實際應用場景

場景 為何需要 Lifespan 典型做法
機器學習模型服務 大型模型載入耗時，需要在第一次請求前完成 在 startup 中使用 torch.load 或 tf.keras.models.load_model，將模型存於 app.state.model；在 shutdown 釋放 GPU 記憶體。
多租戶資料庫 每個租戶有不同的資料庫連線池，必須在服務啟動時預先建立或延遲建立 在 startup 建立 共用 連線池，使用依賴注入根據租戶 ID 動態取得對應的 pool；在 shutdown 一次性關閉所有 pool。
訊息佇列（RabbitMQ、Kafka） 消費者需要持續監聽佇列，服務關閉時必須優雅停止 在 startup 建立 AsyncIO 背景任務（app.add_event_handler("startup", start_consumer)），在 shutdown 呼叫 consumer.stop()。
定時任務（APScheduler、Celery Beat） 任務排程器必須在應用啟動時啟動，關閉時停止 在 startup 呼叫 scheduler.start()，在 shutdown 呼叫 scheduler.shutdown(wait=False)。
Docker / Kubernetes 滾動升級 容器關閉前需要完成資料同步或寫入 finalizer 在 shutdown 中執行 await flush_metrics()、await close_file_handles()，確保容器退出前不遺失資料。

總結

FastAPI 的 lifespan events（startup / shutdown） 為開發者提供了在應用程式生命週期關鍵時刻執行初始化與清理工作的官方機制。透過 @app.on_event 或 lifespan context manager，我們可以：

  • 一次性載入資源（資料庫、快取、模型、設定檔），避免在每個請求中重複建立，提升效能。
  • 安全釋放資源，防止記憶體洩漏或連線遺留，確保服務在容器化環境中能平滑關閉。
  • 以 app.state 為中心，提供全局共享且易於測試的資源存取方式。
  • 結合依賴注入，讓路由函式保持乾淨，同時仍能取得已初始化的資源。

在實務開發中，遵守「初始化 → 使用 → 釋放」的資源管理循環，配合適當的日誌、重試與監控，能讓 FastAPI 應用在高可用、可擴展的環境中穩定運行。希望本篇文章能幫助你在專案中正確運用 Lifespan Events，打造更可靠的 API 服務。祝開發順利！