luk

0001-01-01 約 1300 字預計閱讀 6 分鐘

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

📦 安裝

# 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	查詢建構器

🏗️ 連線設定

同步連線

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()

非同步連線

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

# 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

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 風格（推薦）

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）

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

批量建立

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）

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()

進階查詢

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）

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）

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

# 自動關閉 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 依賴注入

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

📝 建立資料表

# 建立所有資料表
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 風格

# 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），它：
追蹤物件狀態：新增、修改、刪除
管理交易：commit() 和 rollback()
提供查詢介面：query() 和 execute()
物件身份映射：確保同一筆資料只有一個物件實例
with Session(engine) as session:
    user = session.get(User, 1)  # 查詢
    user.name = "New Name"        # 追蹤變更
    session.commit()              # 提交交易

🤓 小測驗

如何建立非同步的 SQLAlchemy Engine？
db.refresh(obj) 的作用是什麼？
SQLAlchemy 2.0 推薦用什麼方式查詢？

上一篇： 02-7. Pydantic v1 vs v2 遷移指南 下一篇： 03-2. 關聯關係

最後更新：2025-12-17