Python 課程 – 資料庫操作（Databases）

主題：ORM（SQLAlchemy）

---

簡介

在 Python 開發中，資料庫操作是日常工作不可或缺的一環。直接使用 SQL 雖然靈活，但對於大型專案而言，維護成本高、容易產生 SQL 注入等安全問題。Object‑Relational Mapping（簡稱 ORM）提供了「物件」與「資料表」之間的映射，讓開發者可以以 Python 類別的方式操作資料，而不必手寫繁雜的 SQL 語句。

SQLAlchemy 是目前 Python 生態系中最受歡迎、功能最完整的 ORM 套件。它同時支援 Core（低階 SQL 表達）與 ORM（高階物件映射），讓你可以根據需求靈活選擇。掌握 SQLAlchemy 後，你將能：

• 快速建構資料模型，提升開發效率。
• 保持資料庫獨立性，未來換 DB 時只需調整連線設定。
• 減少錯誤，因為 ORM 會自動處理參數化查詢，降低 SQL 注入風險。

以下內容將從概念說明、實作範例、常見陷阱與最佳實踐，帶你一步步熟悉 SQLAlchemy 的使用方式。

---

核心概念

1. Engine、Session 與 Base

名稱	功能	常見變數名稱
Engine	負責與資料庫建立連線與執行 SQL。	engine
Session	工作單位（unit of work），負責追蹤物件變化、提交或回滾交易。	Session
Base	所有 ORM 類別的基底，提供 __tablename__、欄位映射等功能。	Base

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

# 建立 Engine（此處使用 SQLite 作為示範）
engine = create_engine('sqlite:///example.db', echo=True)

# 建立 SessionFactory
Session = sessionmaker(bind=engine)

# 建立 Base 類別，所有模型都繼承它
Base = declarative_base()

2. 定義模型（Model）

模型本質上是 Python 類別，每個屬性對應資料表的欄位。使用 Column、Integer、String 等型別進行宣告。

from sqlalchemy import Column, Integer, String, DateTime, func

class User(Base):
    """使用者資料表模型"""
    __tablename__ = 'users'          # 資料表名稱

    id = Column(Integer, primary_key=True)   # 主鍵
    name = Column(String(50), nullable=False)   # 使用者名稱
    email = Column(String(120), unique=True)    # 電子郵件（唯一）
    created_at = Column(DateTime, server_default=func.now())  # 建立時間

Tip：__tablename__ 必須明確指定，否則 SQLAlchemy 會自動產生（但不易掌控）。

3. 建立資料表

使用 Base.metadata.create_all() 讓 SQLAlchemy 依照模型自動產生資料表。

# 只在第一次執行時呼叫，之後若有變更請使用 migration 工具（如 Alembic）
Base.metadata.create_all(engine)

4. CRUD 操作

(1) 新增（Create）

# 建立 Session 實例
session = Session()

# 建立新使用者物件
new_user = User(name='王小明', email='xiaoming@example.com')

# 加入 Session，並提交交易
session.add(new_user)
session.commit()          # 若出錯會自動 rollback
session.close()

(2) 讀取（Read）

session = Session()

# 以主鍵查詢
user = session.get(User, 1)          # 等同於 SELECT * FROM users WHERE id=1
print(user.name, user.email)

# 使用 filter 條件查詢多筆資料
users = session.query(User).filter(User.name.like('%王%')).all()
for u in users:
    print(u.id, u.name)

session.close()

(3) 更新（Update）

session = Session()

# 取得要更新的物件
user = session.get(User, 1)
user.email = 'new_email@example.com'   # 直接修改屬性即可

session.commit()   # ORM 會自動產生 UPDATE 語句
session.close()

(4) 刪除（Delete）

session = Session()

# 取得要刪除的物件
user = session.get(User, 1)
session.delete(user)   # 標記為刪除

session.commit()
session.close()

5. 關聯（Relationship）與外鍵

SQLAlchemy 允許一對多、多對多等關聯的定義。以下示範「使用者」與「文章」的一對多關係。

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Article(Base):
    __tablename__ = 'articles'

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))   # 外鍵指向 users.id

    # 設定反向關聯，使 User 物件可直接取得其所有文章
    author = relationship('User', back_populates='articles')

# 在 User 模型中加入反向屬性
User.articles = relationship('Article', back_populates='author', cascade='all, delete-orphan')

# 新增一篇文章給特定使用者
session = Session()
user = session.get(User, 1)

article = Article(title='SQLAlchemy 入門', content='...', author=user)
session.add(article)
session.commit()
session.close()

---

常見陷阱與最佳實踐

陷阱	說明	解決方式
Session 共享	多個請求共用同一個 Session 會導致資料不一致與記憶體泄漏。	為每一次請求（或每個執行緒）建立獨立 Session，使用 session.close() 或 with Session() as s: 方式確保釋放。
忘記 commit	只呼叫 add() 而未 commit()，資料不會寫入資料庫。	務必 在完成變更後呼叫 session.commit()，或使用 session.flush() 先同步。
懶加載（lazy loading）過度	預設關聯會在存取時才發送額外查詢，若大量迭代會造成 N+1 問題。	使用 joinedload() 或 selectinload() 於查詢時提前載入關聯資料。
使用 SQLite 在多執行緒環境	SQLite 預設不支援多執行緒寫入，會拋出 sqlite3.ProgrammingError。	在 create_engine 時加上 connect_args={'check_same_thread': False}，或改用支援多執行緒的資料庫（PostgreSQL、MySQL）。
模型變更未同步	直接修改模型後忘記執行 migration，導致程式與資料庫結構不一致。	使用 Alembic 產生 migration 檔，並在部署時執行。

最佳實踐：

1. 使用 declarative_base() 統一管理模型，避免混用舊式映射方式。
2. 把 Session 的生命週期交給框架（如 Flask‑SQLAlchemy、FastAPI 的依賴注入），確保每次請求自動開關。
3. 啟用 echo=False 在正式環境關閉 SQL 輸出，減少 I/O 開銷。
4. 對常用查詢加上索引（Index、unique=True），提升效能。
5. 使用 typing（如 from typing import List）為模型加入型別提示，提升 IDE 補完與可讀性。

---

實際應用場景

場景	為何選擇 SQLAlchemy ORM
Web 後端 API（Flask、FastAPI）	以模型作為資料傳輸物件，搭配 Pydantic 可直接序列化/反序列化。
資料分析平台	透過 ORM 抽象化資料來源，讓分析程式碼只關注業務邏輯，而非資料庫結構。
微服務間共享資料模型	多個服務共用同一套模型檔案，確保資料結構一致性。
遺留系統升級	先以 Core 寫低階查詢，逐步遷移至 ORM，降低一次性改寫風險。
測試環境	使用 SQLite 記憶體模式 (sqlite:///:memory:) 快速建立測試資料庫，配合 session.rollback() 保持測試隔離。

範例：在 FastAPI 中結合 SQLAlchemy 與 Pydantic，實作一個簡易的使用者 CRUD API。

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel, EmailStr
from typing import List

app = FastAPI()
# 省略 engine、Session、Base、User 定義（同前）

# Pydantic schema
class UserCreate(BaseModel):
    name: str
    email: EmailStr

class UserRead(BaseModel):
    id: int
    name: str
    email: EmailStr

    class Config:
        orm_mode = True

# 依賴注入 Session
def get_db():
    db = Session()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/", response_model=UserRead)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)   # 取得自動產生的 id
    return db_user

@app.get("/users/", response_model=List[UserRead])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = db.query(User).offset(skip).limit(limit).all()
    return users

這段程式展示了 ORM 與 API 框架 的自然結合：模型直接映射資料表，Pydantic 負責資料驗證與序列化，開發者只需要關注業務流程。

---

總結

SQLAlchemy 作為 Python 生態系最成熟的 ORM，提供了 Engine、Session、Model 三大基礎概念，讓開發者能以物件導向的方式安全、快速地操作資料庫。掌握以下要點，即可在實務專案中得心應手：

1. 正確建立 Engine 與 Session，確保連線與交易管理安全。
2. 使用 Declarative Base 定義模型，並善用 Column、relationship 建立欄位與關聯。
3. 熟悉 CRUD 操作，包括 add()、commit()、query()、delete() 等常用方法。
4. 避免常見陷阱（Session 共享、懶加載 N+1、資料庫遷移）並遵循最佳實踐。
5. 將 ORM 與框架結合（Flask、FastAPI 等），提升開發效率與程式可讀性。

透過本文的概念說明與實作範例，你應該已具備在 Python 專案中導入 SQLAlchemy ORM 的基礎能力。未來可進一步探索 Alembic 進行資料庫遷移、Hybrid Property、自訂型別 等進階功能，讓資料層的設計更具彈性與可維護性。祝你在資料庫操作的道路上越走越順！