# 

# 03-1. SQLAlchemy 基礎

> ⏱️ **閱讀時間：** 18 分鐘
> 🎯 **難度：** ⭐⭐ (基礎)

---

## 🤔 一句話解釋

**SQLAlchemy 是 Python 最強大的 ORM，讓你用 Python 物件操作資料庫，不用寫 SQL。**

---

## 🎯 為什麼選擇 SQLAlchemy？

### ORM vs 原生 SQL

```
❌ 原生 SQL：
cursor.execute("SELECT * FROM users WHERE age > 18 AND status = 'active'")

✅ SQLAlchemy ORM：
session.query(User).filter(User.age > 18, User.status == "active")
```

### 主要優勢

| 優勢 | 說明 |
|------|------|
| **型別安全** | IDE 自動補全，減少錯誤 |
| **SQL 注入防護** | 自動參數化查詢 |
| **跨資料庫** | 換資料庫只需改設定 |
| **關聯管理** | 自動處理 JOIN 和 Foreign Key |
| **遷移支援** | 配合 Alembic 管理 Schema |

---

## 📦 安裝

```bash
# SQLAlchemy 2.0（推薦）
pip install sqlalchemy

# 資料庫驅動程式
pip install psycopg2-binary    # PostgreSQL
pip install pymysql            # MySQL
pip install aiosqlite          # SQLite（非同步）

# 非同步支援
pip install sqlalchemy[asyncio]
```

---

## 🔧 基本概念

### SQLAlchemy 架構

```
┌─────────────────────────────────────────────────────────┐
│                    SQLAlchemy                           │
├─────────────────────────────────────────────────────────┤
│  ORM Layer（物件關聯映射）                                │
│  ┌─────────────────────────────────────────────────┐   │
│  │  Session    Model    Query    Relationship      │   │
│  └─────────────────────────────────────────────────┘   │
├─────────────────────────────────────────────────────────┤
│  Core Layer（SQL 表達式語言）                            │
│  ┌─────────────────────────────────────────────────┐   │
│  │  Engine    Connection    Table    Column        │   │
│  └─────────────────────────────────────────────────┘   │
├─────────────────────────────────────────────────────────┤
│  DBAPI Layer（資料庫介面）                               │
│  ┌─────────────────────────────────────────────────┐   │
│  │  psycopg2    pymysql    sqlite3    ...          │   │
│  └─────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────┘
```

### 核心元件

| 元件 | 說明 |
|------|------|
| **Engine** | 資料庫連線池，管理連線 |
| **Session** | 工作單元，管理物件狀態和交易 |
| **Model** | 對應資料表的 Python 類別 |
| **Query** | 查詢建構器 |

---

## 🏗️ 連線設定

### 同步連線

```python
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 建立 Engine
# 格式: dialect+driver://username:password@host:port/database
DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"

engine = create_engine(
    DATABASE_URL,
    echo=True,           # 印出 SQL（開發用）
    pool_size=5,         # 連線池大小
    max_overflow=10,     # 額外連線數
    pool_pre_ping=True   # 連線前檢查
)

# 建立 Session 工廠
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 建立基礎類別
Base = declarative_base()
```

### 非同步連線

```python
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base

# 非同步 Engine
DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"

async_engine = create_async_engine(
    DATABASE_URL,
    echo=True,
    pool_size=5,
    max_overflow=10
)

# 非同步 Session 工廠
AsyncSessionLocal = sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False
)

Base = declarative_base()
```

### 常見資料庫 URL

```python
# PostgreSQL
"postgresql://user:pass@localhost:5432/dbname"
"postgresql+asyncpg://user:pass@localhost:5432/dbname"  # 非同步

# MySQL
"mysql+pymysql://user:pass@localhost:3306/dbname"
"mysql+aiomysql://user:pass@localhost:3306/dbname"  # 非同步

# SQLite
"sqlite:///./app.db"
"sqlite+aiosqlite:///./app.db"  # 非同步

# SQLite 記憶體
"sqlite:///:memory:"
```

---

## 📝 定義 Model

### 基本 Model

```python
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Text
from sqlalchemy.sql import func
from datetime import datetime

class User(Base):
    __tablename__ = "users"

    # 主鍵
    id = Column(Integer, primary_key=True, index=True)

    # 字串欄位
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(100), unique=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)

    # 文字欄位（長文本）
    bio = Column(Text, nullable=True)

    # 布林欄位
    is_active = Column(Boolean, default=True)
    is_superuser = Column(Boolean, default=False)

    # 時間戳記
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # 使用 server_default（資料庫層級預設值）
    # created_at = Column(DateTime, server_default=func.now())

    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}')>"
```

### SQLAlchemy 2.0 風格（推薦）

```python
from sqlalchemy import String, Boolean, DateTime, Text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
from typing import Optional

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    # 使用 Mapped 型別提示
    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(100), unique=True)
    hashed_password: Mapped[str] = mapped_column(String(255))

    # Optional 表示可為 None
    bio: Mapped[Optional[str]] = mapped_column(Text, nullable=True)

    # 預設值
    is_active: Mapped[bool] = mapped_column(default=True)
    is_superuser: Mapped[bool] = mapped_column(default=False)

    # 時間戳記
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    updated_at: Mapped[Optional[datetime]] = mapped_column(
        onupdate=datetime.utcnow,
        nullable=True
    )

    def __repr__(self) -> str:
        return f"<User(id={self.id}, username='{self.username}')>"
```

### 常用欄位型別

| SQLAlchemy 型別 | Python 型別 | 說明 |
|-----------------|-------------|------|
| `Integer` | `int` | 整數 |
| `BigInteger` | `int` | 大整數 |
| `String(n)` | `str` | 固定長度字串 |
| `Text` | `str` | 長文本 |
| `Boolean` | `bool` | 布林值 |
| `Float` | `float` | 浮點數 |
| `Numeric(p, s)` | `Decimal` | 精確數字 |
| `Date` | `date` | 日期 |
| `DateTime` | `datetime` | 日期時間 |
| `Time` | `time` | 時間 |
| `JSON` | `dict/list` | JSON 資料 |
| `LargeBinary` | `bytes` | 二進位資料 |
| `Enum` | `Enum` | 列舉 |
| `UUID` | `uuid.UUID` | UUID |

---

## 🔍 CRUD 操作

### 建立（Create）

```python
from sqlalchemy.orm import Session

def create_user(db: Session, username: str, email: str, password: str) -> User:
    # 建立物件
    user = User(
        username=username,
        email=email,
        hashed_password=password  # 實際應該雜湊
    )

    # 加入 Session
    db.add(user)

    # 提交交易
    db.commit()

    # 重新載入（取得自動生成的 id）
    db.refresh(user)

    return user

# 使用
with SessionLocal() as db:
    user = create_user(db, "john", "john@example.com", "hashed_pw")
    print(user.id)  # 自動生成的 ID
```

### 批量建立

```python
def create_users_bulk(db: Session, users_data: list[dict]) -> list[User]:
    users = [User(**data) for data in users_data]
    db.add_all(users)
    db.commit()

    # 重新載入所有物件
    for user in users:
        db.refresh(user)

    return users
```

### 查詢（Read）

```python
from sqlalchemy import select
from sqlalchemy.orm import Session

# ===== SQLAlchemy 2.0 風格（推薦）=====

def get_user_by_id(db: Session, user_id: int) -> User | None:
    """根據 ID 查詢"""
    stmt = select(User).where(User.id == user_id)
    return db.scalar(stmt)

def get_user_by_email(db: Session, email: str) -> User | None:
    """根據 email 查詢"""
    stmt = select(User).where(User.email == email)
    return db.scalar(stmt)

def get_users(
    db: Session,
    skip: int = 0,
    limit: int = 100
) -> list[User]:
    """查詢多個使用者（分頁）"""
    stmt = select(User).offset(skip).limit(limit)
    return list(db.scalars(stmt).all())

def get_active_users(db: Session) -> list[User]:
    """查詢啟用的使用者"""
    stmt = select(User).where(User.is_active == True)
    return list(db.scalars(stmt).all())

# ===== SQLAlchemy 1.x 風格（仍可使用）=====

def get_user_by_id_legacy(db: Session, user_id: int) -> User | None:
    return db.query(User).filter(User.id == user_id).first()

def get_users_legacy(db: Session, skip: int = 0, limit: int = 100) -> list[User]:
    return db.query(User).offset(skip).limit(limit).all()
```

### 進階查詢

```python
from sqlalchemy import select, and_, or_, desc, asc, func

def search_users(
    db: Session,
    *,
    username: str | None = None,
    email: str | None = None,
    is_active: bool | None = None,
    order_by: str = "created_at",
    order_desc: bool = True,
    skip: int = 0,
    limit: int = 100
) -> list[User]:
    """進階搜尋"""
    stmt = select(User)

    # 動態條件
    conditions = []
    if username:
        conditions.append(User.username.ilike(f"%{username}%"))
    if email:
        conditions.append(User.email.ilike(f"%{email}%"))
    if is_active is not None:
        conditions.append(User.is_active == is_active)

    if conditions:
        stmt = stmt.where(and_(*conditions))

    # 排序
    order_column = getattr(User, order_by, User.created_at)
    if order_desc:
        stmt = stmt.order_by(desc(order_column))
    else:
        stmt = stmt.order_by(asc(order_column))

    # 分頁
    stmt = stmt.offset(skip).limit(limit)

    return list(db.scalars(stmt).all())

def count_users(db: Session, is_active: bool | None = None) -> int:
    """計算使用者數量"""
    stmt = select(func.count()).select_from(User)
    if is_active is not None:
        stmt = stmt.where(User.is_active == is_active)
    return db.scalar(stmt) or 0
```

### 更新（Update）

```python
from sqlalchemy import update

def update_user(
    db: Session,
    user_id: int,
    **kwargs
) -> User | None:
    """更新使用者"""
    # 方法 1: 查詢後更新（會觸發 ORM 事件）
    user = db.get(User, user_id)
    if not user:
        return None

    for key, value in kwargs.items():
        if hasattr(user, key):
            setattr(user, key, value)

    db.commit()
    db.refresh(user)
    return user

def update_user_bulk(
    db: Session,
    user_ids: list[int],
    **kwargs
) -> int:
    """批量更新"""
    # 方法 2: 直接 UPDATE（不觸發 ORM 事件，效能更好）
    stmt = (
        update(User)
        .where(User.id.in_(user_ids))
        .values(**kwargs)
    )
    result = db.execute(stmt)
    db.commit()
    return result.rowcount

def deactivate_user(db: Session, user_id: int) -> bool:
    """停用使用者"""
    user = update_user(db, user_id, is_active=False)
    return user is not None
```

### 刪除（Delete）

```python
from sqlalchemy import delete

def delete_user(db: Session, user_id: int) -> bool:
    """刪除使用者"""
    # 方法 1: 查詢後刪除
    user = db.get(User, user_id)
    if not user:
        return False

    db.delete(user)
    db.commit()
    return True

def delete_users_bulk(db: Session, user_ids: list[int]) -> int:
    """批量刪除"""
    # 方法 2: 直接 DELETE
    stmt = delete(User).where(User.id.in_(user_ids))
    result = db.execute(stmt)
    db.commit()
    return result.rowcount

def soft_delete_user(db: Session, user_id: int) -> bool:
    """軟刪除（標記為刪除）"""
    return update_user(db, user_id, is_deleted=True) is not None
```

---

## 🔄 Session 管理

### 使用 Context Manager

```python
# 自動關閉 Session
with SessionLocal() as db:
    users = get_users(db)
    print(users)
# Session 自動關閉

# 搭配 try/except
with SessionLocal() as db:
    try:
        create_user(db, "john", "john@example.com", "password")
    except Exception:
        db.rollback()
        raise
```

### FastAPI 依賴注入

```python
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

# 依賴項
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users")
def read_users(db: Session = Depends(get_db)):
    return get_users(db)

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = get_user_by_id(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user
```

---

## 📝 建立資料表

```python
# 建立所有資料表
Base.metadata.create_all(bind=engine)

# 刪除所有資料表
Base.metadata.drop_all(bind=engine)

# 只建立特定資料表
User.__table__.create(bind=engine)
```

---

## ✅ 重點總結

### SQLAlchemy 核心概念

| 概念 | 說明 |
|------|------|
| **Engine** | 資料庫連線池 |
| **Session** | 工作單元，管理交易 |
| **Model** | 對應資料表的類別 |
| **Query/Select** | 查詢建構 |

### CRUD 方法對照

| 操作 | 方法 |
|------|------|
| **Create** | `db.add()`, `db.commit()` |
| **Read** | `db.scalar(select(...))`, `db.get()` |
| **Update** | 修改屬性 + `db.commit()` |
| **Delete** | `db.delete()`, `db.commit()` |

### 2.0 vs 1.x 風格

```python
# 1.x 風格
db.query(User).filter(User.id == 1).first()

# 2.0 風格（推薦）
db.scalar(select(User).where(User.id == 1))
```

---

## 🎤 面試這樣答

### Q: SQLAlchemy 的 Session 是什麼？

**答案：**

> Session 是 SQLAlchemy 的工作單元（Unit of Work），它：
>
> 1. **追蹤物件狀態**：新增、修改、刪除
> 2. **管理交易**：commit() 和 rollback()
> 3. **提供查詢介面**：query() 和 execute()
> 4. **物件身份映射**：確保同一筆資料只有一個物件實例
>
> ```python
> with Session(engine) as session:
>     user = session.get(User, 1)  # 查詢
>     user.name = "New Name"        # 追蹤變更
>     session.commit()              # 提交交易
> ```

---

## 🤓 小測驗

1. 如何建立非同步的 SQLAlchemy Engine？
   <details>
   <summary>點擊看答案</summary>
   使用 `create_async_engine()` 和非同步驅動程式（如 asyncpg）
   </details>

2. `db.refresh(obj)` 的作用是什麼？
   <details>
   <summary>點擊看答案</summary>
   從資料庫重新載入物件的屬性（例如取得自動生成的 ID）
   </details>

3. SQLAlchemy 2.0 推薦用什麼方式查詢？
   <details>
   <summary>點擊看答案</summary>
   使用 `select()` 語句：`db.scalar(select(User).where(...))`
   </details>

---

**上一篇：** [02-7. Pydantic v1 vs v2 遷移指南](./02-7)
**下一篇：** [03-2. 關聯關係](./03-2)

---

最後更新：2025-12-17


---

> 作者: luk  
> URL: https://yoru-karu-blog-lalaluk-52581ac5e0cef170a3c8922c19182ecb6f7bd604.gitlab.io/posts/tutorial/fastapi/03-1/