在前面的章節中，我們學習了 FastAPI 的核心概念和非同步程式設計的威力。今天，我們要將這些知識應用到實際開發中最重要的環節之一——資料庫操作。

我們將探討如何在 FastAPI 中建立高效的資料庫連線，以及為什麼非同步資料庫操作是發揮 FastAPI 最大效能的關鍵。

為什麼要在 FastAPI 中使用「非同步」操作資料庫？

你可能會想，我以前用 Flask 或 Django，直接用 pymysql 或 psycopg2 這些標準函式庫就能連線資料庫，為什麼在 FastAPI 中需要特別強調「非同步」？

這要回到我們在 Day 03 提到的觀念：同步 vs. 非同步。

傳統的資料庫操作是一個典型的 I/O bound (輸出/入密集型) 任務。當我們的程式向資料庫發出一個查詢請求時，在資料庫處理查詢並返回結果的這段時間裡，我們的程式其實是在「等待」。

• 在同步的世界裡 (如 WSGI 框架)，這個等待會阻塞 (Blocking) 住整個工作執行緒。如果同時有 100 個請求進來，就需要 100 個執行緒來處理，這會消耗大量記憶體資源。
• 在非同步的世界裡 (如 FastAPI)，事件迴圈 (Event Loop) 可以在發出 DB 查詢後，立刻轉身去處理其他請求，完全不會被「等待」這個動作卡住。等到資料庫回傳結果時，事件迴圈再回來處理後續的邏輯。

這就是 FastAPI 搭配非同步資料庫操作能實現超高效能的關鍵。我們用更少的資源，就能處理更高的併發請求。

選擇你的非同步工具

要實現非同步資料庫操作，我們需要兩樣東西：

1. 一個支援 asyncio 的資料庫驅動 (Driver)。
2. 一個支援非同步的 ORM (Object-Relational Mapper) 或查詢建構器 (Query Builder)。

以常見的 PostgreSQL 資料庫為例，一個非常主流的組合是：

• 資料庫驅動: asyncpg - 一個專為 asyncio 設計，效能極高的 PostgreSQL 驅動程式。
• ORM: SQLAlchemy - 從 1.4 版開始，SQLAlchemy 提供了完整的 asyncio 支援，讓我們可以用熟悉的語法進行非同步操作。

基礎範例：連線與查詢

讓我們先看一個最基礎的連線與查詢範例。

首先，安裝必要的套件：

pip install "sqlalchemy[asyncio]" asyncpg

接下來，這是一個在 FastAPI 中設定非同步資料庫連線的簡單範例：

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text
from fastapi import FastAPI, Depends
from typing import AsyncGenerator

# 記得替換成你自己的資料庫連線資訊
DATABASE_URL = "postgresql+asyncpg://user:password@host/dbname"

# 建立非同步的 Engine
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionFactory = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

app = FastAPI()

# 使用 FastAPI 的 Dependency Injection 來取得 DB Session
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionFactory() as session:
        yield session

@app.get("/db-test")
async def test_db_connection(db: AsyncSession = Depends(get_db)):
    # SQLAlchemy 2.0 語法
    # 執行一個簡單的 SQL 查詢來測試連線
    result = await db.execute(text("SELECT 1"))
    value = result.scalar_one()
    return {"message": "Database connection successful!", "value": value}

在這個範例中，我們做了幾件重要的事：

1. 使用 create_async_engine 建立了一個 SQLAlchemy 的非同步引擎。

2. 定義了一個 get_db 的非同步生成器 (AsyncGenerator)，這是 FastAPI 中管理每個請求的資料庫連線的標準模式。

3. 在路徑操作函式中，透過 Depends(get_db) 注入 AsyncSession，並在查詢時使用 await 關鍵字。

小結

今天我們學習了 FastAPI 中非同步資料庫操作的核心概念：使用 create_async_engine 建立引擎、透過 Depends 注入 AsyncSession，以及為什麼資料庫這類 I/O bound 操作適合使用非同步處理。

掌握了這些基礎後，我們就能在 FastAPI 中建立高效能的資料庫應用。

資料庫的部分應該就先這樣了，之後有機會的話再展開介紹，明天會開始介紹任務管理。