Python SQLAlchemy 2.0 實戰：Typed ORM、Session 與查詢模式

Python 學習 - 本文屬於一個選集。

§ 66: 本文

一. 前言：資料庫程式最怕「看起來能跑」
#

Python 寫資料庫程式，很容易從「先用字串組 SQL」開始，然後慢慢變成每個 function 都知道資料表欄位長什麼樣子。剛開始很快，但欄位改名、transaction 邊界、測試資料、API schema 一起出現時，維護成本就會開始追殺你。如果你只是寫一次性腳本，Python 內建的 sqlite3 很夠用。可是如果你在寫會長大的 CLI、背景工作、FastAPI 服務，或任何需要穩定資料存取層的工具，拍拍君會建議你認真認識 SQLAlchemy 2.0。 SQLAlchemy 不是要你忘記 SQL。更好的理解是：它讓資料表、查詢、transaction 與資料庫連線，用一套清楚的 Python 物件模型管理起來。今天我們做一個小型書籤資料庫，會看到：

typed ORM model 怎麼寫
Engine 與 Session 怎麼分工
CRUD 怎麼切得乾淨
select 查詢怎麼用
relationship 與載入策略要注意什麼
repository pattern 什麼時候值得用

先把核心觀念放在桌上：

Model 描述資料，Session 管理變更，外層決定 transaction。這三件事分清楚，資料庫程式就會穩很多。

二. 安裝：建立乾淨專案
#

這篇用 SQLite 示範，因為它不需要另外啟動資料庫服務。換成 PostgreSQL 時，概念大致相同，只是連線字串與 driver 不同。

mkdir sqlalchemy-demo
cd sqlalchemy-demo
uv init
uv add sqlalchemy

如果你不用 uv，也可以用一般 venv：

python -m venv .venv
source .venv/bin/activate
python -m pip install sqlalchemy

確認版本：

python - <<'PY'
import sqlalchemy
print(sqlalchemy.__version__)
PY

本文使用 SQLAlchemy 2.x 寫法：

Mapped[T]
mapped_column()
select(Model)
Session(engine)
不用舊式 session.query(Model) 當主要範例

專案結構：

sqlalchemy-demo/
├── bookmark_app/
│   ├── __init__.py
│   ├── db.py
│   ├── models.py
│   └── repository.py
└── main.py

建立檔案：

mkdir bookmark_app
touch bookmark_app/__init__.py
touch bookmark_app/db.py bookmark_app/models.py bookmark_app/repository.py
touch main.py

三. Engine 與 Base：資料庫入口
#

SQLAlchemy 第一個重要物件是 Engine。你可以把它想成資料庫連線池與 SQL 執行能力的入口。在 bookmark_app/db.py：

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


engine = create_engine(
    "sqlite:///bookmarks.db",
    echo=False,
)

sqlite:///bookmarks.db 代表使用目前目錄下的 bookmarks.db。初學時可以把 echo=True 打開，觀察 SQLAlchemy 實際送出的 SQL；正式程式通常不要一直開著，log 會很吵。 Base 是所有 ORM model 的共同基底。等一下每個資料表 class 都會繼承它。

四. 第一個 Model：用 class 描述資料表
#

在 bookmark_app/models.py 寫一個書籤資料表：

from datetime import datetime

from sqlalchemy import String, Text
from sqlalchemy.orm import Mapped, mapped_column

from .db import Base


class Bookmark(Base):
    __tablename__ = "bookmarks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(120))
    url: Mapped[str] = mapped_column(String(500), unique=True)
    note: Mapped[str | None] = mapped_column(Text, default=None)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

Mapped[str] 不是裝飾品。它讓型別檢查器與 IDE 知道：這是 ORM 欄位，而從資料庫拿出的 bookmark.title 是 str。 mapped_column() 負責欄位設定：

primary_key=True：主鍵
String(120)：字串長度
unique=True：唯一限制
default=datetime.utcnow：建立資料時自動填入

注意最後一個是 datetime.utcnow，不是 datetime.utcnow()。前者把 function 交給 SQLAlchemy，後者會在 import 時立刻執行，容易讓多筆資料拿到同一個時間。這種小坑很無聊，但真的會咬人。

五. 建表：metadata.create_all
#

Model 有了，但資料庫還沒有表。先在 main.py 初始化：

from bookmark_app.db import Base, engine
from bookmark_app.models import Bookmark


def main() -> None:
    Base.metadata.create_all(engine)
    print("database ready")


if __name__ == "__main__":
    main()

執行：

python main.py

你會看到專案裡多一個 bookmarks.db。 Base.metadata.create_all(engine) 很適合教學、小工具、測試資料庫與原型開發。但正式產品通常會搭配 Alembic 做 migration，因為 create_all 擅長「從無到有」，不擅長「欄位改名、型別調整、資料搬移」。

六. Session：transaction 的工作區
#

SQLAlchemy 第二個關鍵物件是 Session。很多新手會把 Session 當成連線，但比較實用的理解是：

Session 是一個資料庫工作區，負責追蹤物件變更，最後一起 flush / commit。先新增一筆資料：

from sqlalchemy.orm import Session

from bookmark_app.db import Base, engine
from bookmark_app.models import Bookmark


def main() -> None:
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        bookmark = Bookmark(
            title="Daily Pypy",
            url="https://dailypypy.org/",
            note="拍拍君的技術筆記",
        )
        session.add(bookmark)
        session.commit()

        print(bookmark.id)


if __name__ == "__main__":
    main()

這段程式建立 Bookmark 物件，加入 Session 追蹤，最後 commit() 寫入資料庫。commit 後，bookmark.id 會拿到資料庫產生的主鍵。 Session 的生命週期要短：

CLI：一個 command 一個 Session
Web request：一個 request 一個 Session
背景工作：一個 job 一個 Session
測試：一個 test 或 fixture 一個 Session

Engine 可以共用。Session 不要亂共用。拍拍君在這裡很囉嗦，因為這是很多資料庫 bug 的起點。

七. 查詢：用 select，不用舊式 query
#

SQLAlchemy 2.0 推薦使用 select()。

from sqlalchemy import select
from sqlalchemy.orm import Session

from bookmark_app.db import engine
from bookmark_app.models import Bookmark


with Session(engine) as session:
    statement = select(Bookmark).order_by(Bookmark.created_at.desc())
    bookmarks = session.scalars(statement).all()

    for bookmark in bookmarks:
        print(bookmark.title, bookmark.url)

session.scalars(statement) 的意思是：我想拿每列的主要 ORM 物件。查單筆可以這樣寫：

statement = select(Bookmark).where(Bookmark.url == "https://dailypypy.org/")
bookmark = session.scalars(statement).one_or_none()

常用取值方法：

all()：全部拿出來
first()：第一筆，沒有就 None
one()：必須剛好一筆，否則丟錯
one_or_none()：零筆或一筆可以，多筆丟錯

拍拍君通常偏好 one_or_none()，因為它能抓出「明明應該唯一，結果資料庫裡有多筆」的問題。太快用 first()，有時只是把資料問題藏起來。

八. CRUD：repository 先從 function 開始
#

把常用操作放進 bookmark_app/repository.py。先從 function 開始，不急著建立 class。

from sqlalchemy import select
from sqlalchemy.orm import Session

from .models import Bookmark


def create_bookmark(
    session: Session,
    *,
    title: str,
    url: str,
    note: str | None = None,
) -> Bookmark:
    bookmark = Bookmark(title=title, url=url, note=note)
    session.add(bookmark)
    session.flush()
    return bookmark


def get_bookmark_by_url(session: Session, url: str) -> Bookmark | None:
    statement = select(Bookmark).where(Bookmark.url == url)
    return session.scalars(statement).one_or_none()


def list_bookmarks(session: Session) -> list[Bookmark]:
    statement = select(Bookmark).order_by(Bookmark.created_at.desc())
    return list(session.scalars(statement).all())

這裡故意使用 session.flush()，不是 session.commit()。repository function 不一定應該決定 transaction 何時結束。比較乾淨的分工是：

repository 負責新增、查詢、修改、刪除
外層 use case 或 command 負責 commit / rollback

例如：

from sqlalchemy.orm import Session

from bookmark_app.db import engine
from bookmark_app.repository import create_bookmark, get_bookmark_by_url


def add_if_missing(title: str, url: str) -> None:
    with Session(engine) as session:
        existing = get_bookmark_by_url(session, url)
        if existing is not None:
            print(f"already exists: {existing.title}")
            return

        bookmark = create_bookmark(session, title=title, url=url)
        session.commit()
        print(f"created #{bookmark.id}: {bookmark.title}")

偷偷 commit 是資料庫程式裡很常見的維護災難。除非 function 本身就是完整工作單位，否則 commit 留給外層會比較穩。

九. 更新、刪除與錯誤處理
#

更新資料時，先查出物件，直接改屬性：

def update_note(session: Session, url: str, note: str) -> Bookmark | None:
    bookmark = get_bookmark_by_url(session, url)
    if bookmark is None:
        return None

    bookmark.note = note
    session.flush()
    return bookmark

刪除也類似：

def delete_bookmark(session: Session, url: str) -> bool:
    bookmark = get_bookmark_by_url(session, url)
    if bookmark is None:
        return False

    session.delete(bookmark)
    session.flush()
    return True

SQLAlchemy 會追蹤物件是否被改過。你不需要自己寫 UPDATE bookmarks SET note = ? WHERE url = ?，但你仍然要知道變更什麼時候送到資料庫：

flush()：把目前變更送到資料庫，但 transaction 還沒結束
commit()：提交 transaction
rollback()：回復 transaction 裡尚未提交的變更

如果 url 設了 unique=True，重複新增會觸發 IntegrityError：

from sqlalchemy.exc import IntegrityError


def safe_create(title: str, url: str) -> None:
    with Session(engine) as session:
        try:
            bookmark = create_bookmark(session, title=title, url=url)
            session.commit()
        except IntegrityError:
            session.rollback()
            print("URL already exists")
            return

        print(f"created #{bookmark.id}")

重點不是把所有錯誤都 catch 起來，而是 transaction 出錯後要 rollback。一個失敗過的 Session，不能假裝沒事繼續用。

十. Relationship：資料表之間要有方向
#

現在加入 tag。為了讓範例不過度膨脹，先做簡化版：每個 tag 屬於一個 bookmark。

from sqlalchemy import ForeignKey, String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship


class Bookmark(Base):
    __tablename__ = "bookmarks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(120))
    url: Mapped[str] = mapped_column(String(500), unique=True)
    note: Mapped[str | None] = mapped_column(Text, default=None)

    tags: Mapped[list["Tag"]] = relationship(
        back_populates="bookmark",
        cascade="all, delete-orphan",
    )


class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), index=True)
    bookmark_id: Mapped[int] = mapped_column(ForeignKey("bookmarks.id"))

    bookmark: Mapped[Bookmark] = relationship(back_populates="tags")

relationship() 讓你可以用 Python 物件操作關聯：

bookmark = Bookmark(
    title="SQLAlchemy",
    url="https://www.sqlalchemy.org/",
    tags=[Tag(name="python"), Tag(name="database")],
)

session.add(bookmark)
session.commit()

cascade="all, delete-orphan" 表示 bookmark 被刪除時，底下的 tag 也一起刪。這很方便，但不要亂加；cascade 是資料生命週期設計，不只是少寫幾行 delete。

十一. 避免 N+1：明確選擇載入策略
#

ORM 最常見的效能坑叫 N+1 query。假設你先查出 20 個 bookmark，接著在迴圈裡讀每個 bookmark 的 tags，如果 tags 是 lazy load，就可能變成 1 次查 bookmarks 加 20 次查 tags。可以用 selectinload：

from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload

from .models import Bookmark


def list_bookmarks_with_tags(session: Session) -> list[Bookmark]:
    statement = (
        select(Bookmark)
        .options(selectinload(Bookmark.tags))
        .order_by(Bookmark.created_at.desc())
    )
    return list(session.scalars(statement).all())

selectinload 通常是很好的預設選項。它會先查主表，再用 IN (...) 查相關資料。不是所有 relationship 都應該 eager load。但當你知道畫面或 API 一定會用到 tags，就明確寫出來。資料庫程式最怕「看起來只是讀一個屬性，背後其實查了一次資料庫」。

十二. Repository Pattern：ORM 不要滲得到處都是
#

小專案裡，直接在 service 寫 SQLAlchemy 查詢沒有問題。專案長大後，可以把資料存取集中在 repository。

from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload

from .models import Bookmark, Tag


class BookmarkRepository:
    def __init__(self, session: Session) -> None:
        self.session = session

    def add(self, title: str, url: str, tags: list[str]) -> Bookmark:
        bookmark = Bookmark(
            title=title,
            url=url,
            tags=[Tag(name=name) for name in tags],
        )
        self.session.add(bookmark)
        self.session.flush()
        return bookmark

    def get_by_url(self, url: str) -> Bookmark | None:
        statement = (
            select(Bookmark)
            .options(selectinload(Bookmark.tags))
            .where(Bookmark.url == url)
        )
        return self.session.scalars(statement).one_or_none()

    def search_by_tag(self, tag_name: str) -> list[Bookmark]:
        statement = (
            select(Bookmark)
            .join(Bookmark.tags)
            .where(Tag.name == tag_name)
            .options(selectinload(Bookmark.tags))
            .order_by(Bookmark.created_at.desc())
        )
        return list(self.session.scalars(statement).unique().all())

這樣外層程式不需要知道 relationship 怎麼 join、tags 要怎麼 eager load、查詢條件怎麼組。外層只要知道它需要「新增書籤」或「用 tag 搜尋」。但不要一開始就把所有東西抽象成 repository。如果專案只有三個查詢，function 很好。等到查詢重複、transaction 變複雜、測試需要隔離時，再抽也不晚。架構是為了降低變更成本，不是為了看起來專業。

十三. 搭配 FastAPI：一個 request 一個 Session
#

如果你在 FastAPI 裡使用 SQLAlchemy，最常見的做法是 dependency。這和 FastAPI 入門的精神一樣：HTTP 邊界與資料庫邊界要分清楚。

from collections.abc import Generator

from sqlalchemy.orm import Session, sessionmaker

from .db import engine


SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)


def get_session() -> Generator[Session, None, None]:
    with SessionLocal() as session:
        yield session

Route 裡注入 Session：

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from bookmark_app.repository import BookmarkRepository
from bookmark_app.session import get_session

app = FastAPI()


@app.get("/bookmarks/{url:path}")
def get_bookmark(url: str, session: Session = Depends(get_session)) -> dict:
    repo = BookmarkRepository(session)
    bookmark = repo.get_by_url(url)
    if bookmark is None:
        raise HTTPException(status_code=404, detail="bookmark not found")

    return {
        "title": bookmark.title,
        "url": bookmark.url,
        "tags": [tag.name for tag in bookmark.tags],
    }

Route 負責 HTTP，repository 負責資料庫，service 或 use case 負責商業規則。不要讓 route 裡塞滿查詢條件，兩週後你自己會看不懂。拍拍君不是在嚇你，是在保護你。

十四. 測試：真的跑過資料庫
#

SQLAlchemy 很適合搭配 pytest。SQLite in-memory 對小型 repository 測試很方便：

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from bookmark_app.db import Base
from bookmark_app.repository import BookmarkRepository


@pytest.fixture
def session() -> Session:
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        yield session


def test_add_and_get_bookmark(session: Session) -> None:
    repo = BookmarkRepository(session)

    created = repo.add(
        title="Daily Pypy",
        url="https://dailypypy.org/",
        tags=["python", "blog"],
    )
    session.commit()

    found = repo.get_by_url("https://dailypypy.org/")

    assert found is not None
    assert found.id == created.id
    assert found.title == "Daily Pypy"
    assert [tag.name for tag in found.tags] == ["python", "blog"]

這種測試比 mock 資料庫更有價值，因為你真的跑到 mapping、constraint、relationship、flush、query。資料庫程式的 bug 常常藏在這些地方。正式專案如果用 PostgreSQL，也應該準備 PostgreSQL 測試環境。SQLite 很方便，但不是所有 SQL 行為都和 PostgreSQL 一樣。

十五. 常見坑：先記住這幾個
#

SQLAlchemy 很強，也因此有些地方不適合靠猜。第一，Session 不要當全域變數到處共用。Engine 可以全域，Session 應該有清楚生命週期。第二，repository 裡不要隨便 commit。除非你的 function 明確就是一個完整 transaction，不然把 commit 留給外層。第三，不要用 ORM 當作不用學 SQL 的藉口。join、index、unique constraint、transaction isolation，這些概念還是資料庫基本功。第四，小心 lazy loading。尤其是 API 回傳列表時，N+1 很容易偷偷長出來。第五，model 不一定等於 API schema。SQLAlchemy model 描述資料庫，Pydantic model 描述 API 契約。它們可以長得很像，但責任不同。第六，先不要追求最漂亮的抽象。先讓 transaction 邊界清楚、查詢可讀、測試能跑，比一開始就設計十層架構重要。