Web Day 8 關聯設計:一對多與多對多
執行需求:CPU 可跑。昨天的書本 API 把作者當作「字串欄位」儲存,這在小規模沒問題,但一旦需要查詢「Hao 這位作者的所有書」、「書本依作者分組」、「同一作者不能重複建立」就會撞牆。今天要進入資料庫設計的核心——關聯(relationship):用 SQLModel 的外鍵與 Relationship 機制,把書本與作者拆成兩張資料表,再示範多對多(書本與標籤)的設計。學完之後,你就具備設計「多資料表」的後端所需的基本功。
引言
把所有欄位塞進同一張表是初學者最常做的事。書本表裡直接存作者字串看起來很方便,但會碰到三個問題:第一,作者名稱要改(例如「王小明」改成「王曉明」)時,必須 UPDATE 所有書本紀錄;第二,沒辦法在「作者」這個層級做事(例如查詢「所有作者」、統計「每位作者的書本數」);第三,作者名稱重複(「Hao」與「Hao 」是不同的字串)會造成同一個作者有多筆紀錄,難以管理。
解決方法是「資料正規化(normalization)」:把重複的資料拆成獨立資料表,並用「外鍵(foreign key)」建立關聯。這就是關聯式資料庫(relational database)的精髓。SQLModel 用 Python class 來表達這些關係:author_id: int | None = Field(foreign_key="author.id") 是外鍵欄位、Relationship() 是物件導向的存取機制。這套機制跟 SQLAlchemy 一致,學一次終身受用。
這篇文章會做四件事:第一,解釋關聯的必要性與正規化基礎;第二,用一對多(一位作者多本書)建立第一個跨資料表關係;第三,示範多對多(書本與標籤)的 join table 設計;第四,建立一個完整的「作者 + 書本 + 標籤」三資料表系統,示範嵌套建立、巢狀回應、cascade 刪除。本篇會延續昨天的書本 API 範例,繼續累積成後續專案篇的基礎。
為什麼要拆資料表:正規化的基礎
資料庫正規化的核心目標是「減少資料冗餘」、「避免更新異常」、「保證資料一致性」。舉個例子,假設我們的書本表現在長這樣:
| id | title | author_name | author_email | year |
|---|---|---|---|---|
| 1 | Python 程式設計實戰 | Hao | hao@example.com | 2024 |
| 2 | 資料庫設計 | Hao | hao@example.com | 2025 |
這個設計的問題:作者 email 重複存了兩次。如果 Hao 想改 email,必須 UPDATE 兩筆;如果只 UPDATE 一筆就會出現「同一位作者有兩個 email」的資料不一致。正規化後拆成兩張表:
| authors | books | ||||
|---|---|---|---|---|---|
| id | name | id | title | author_id (FK) | year |
| 1 | Hao | 1 | Python 程式設計實戰 | 1 | 2024 |
| 2 | 資料庫設計 | 1 | 2025 | ||
現在作者 email 只存一份;Hao 想改 email 只要 UPDATE authors 一筆。要查「Hao 寫的所有書」就是 SELECT * FROM books WHERE author_id = 1,或用 JOIN 一次拿到作者 + 書本資訊。
SQLModel 的一對多關聯
SQLModel 的一對多寫法:用 Field(foreign_key="table.column") 標示外鍵欄位,用 Relationship(back_populates="...") 建立雙向關聯。
# 一對多範例:作者(一)與書本(多)
from typing import TYPE_CHECKING
from sqlmodel import Field, Relationship, SQLModel
if TYPE_CHECKING:
# 避免迴圈匯入;只在型別檢查時匯入
from .book import Book
class Author(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(min_length=1, max_length=80, unique=True, index=True)
email: str = Field(unique=True, index=True)
# 一個作者有多本書
books: list["Book"] = Relationship(back_populates="author")
class Book(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
title: str = Field(min_length=1, max_length=200)
year: int = Field(ge=0, le=2100)
# 外鍵:指向 authors.id
author_id: int | None = Field(default=None, foreign_key="author.id")
# 反向關聯
author: Author | None = Relationship(back_populates="books")
幾個關鍵設計:
author_id: int | None = Field(foreign_key="author.id"):外鍵欄位,可空(書本可以暫時沒有作者)。Relationship(back_populates="books"):在Book這邊告訴 SQLAlchemy「反向關聯到Author.books」。Relationship(back_populates="author"):在Author這邊告訴 SQLAlchemy「反向關聯到Book.author」。TYPE_CHECKING區塊是為了解決「兩個 class 互相 import」造成的迴圈;只在型別檢查時匯入,執行時不會真的 import。
查詢時可以用 SQLModel 的自動載入機制存取關聯:
# 用 SQLModel 走訪關聯
from sqlmodel import Session, select
with Session(engine) as session:
# 查某本書與其作者
book = session.get(Book, 1)
if book and book.author:
print(f"{book.title} 由 {book.author.name} 撰寫")
# 輸出(範例):Python 程式設計實戰 由 Hao 撰寫
# 查某作者的所有書
author = session.get(Author, 1)
if author:
for book in author.books:
print(f" - {book.title}")
# 輸出(範例):
# - Python 程式設計實戰
# - 資料庫設計
多對多:書本與標籤
多對多關係需要第三張表(join table)來記錄「誰對誰」的對應關係。例如一本書可以有多個標籤(Python、FastAPI、Web),一個標籤可以對應多本書。SQLModel 提供了兩種寫法:
- 用
link_model參數自動建立 join table(簡單) - 手動建立 join table class(彈性高,可以加額外欄位)
先用自動寫法:
# 多對多範例:書本與標籤(自動 join table)
from sqlmodel import Field, Relationship, SQLModel
class Tag(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(unique=True, index=True)
# 自動 join table;不需要手動建中間表
books: list["Book"] = Relationship(
back_populates="tags",
link_model="BookTagLink", # 自動產生的中間表 class 名稱
)
class Book(SQLModel, table=True):
# ... 原本的欄位 ...
tags: list[Tag] = Relationship(
back_populates="books",
link_model="BookTagLink",
)
但實務上更常見的是「手動建 join table」,因為這樣可以加額外欄位(例如「加入日期」、「建立者」):
# 手動 join table 範例
from datetime import datetime
from sqlmodel import Field, Relationship, SQLModel
class BookTagLink(SQLModel, table=True):
"""書本與標籤的中間表"""
book_id: int | None = Field(
default=None, foreign_key="book.id", primary_key=True
)
tag_id: int | None = Field(
default=None, foreign_key="tag.id", primary_key=True
)
created_at: datetime | None = Field(default_factory=datetime.utcnow)
class Tag(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(unique=True, index=True)
books: list["Book"] = Relationship(
back_populates="tags",
link_model=BookTagLink,
)
class Book(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
title: str = Field(min_length=1, max_length=200)
year: int = Field(ge=0, le=2100)
tags: list[Tag] = Relationship(
back_populates="books",
link_model=BookTagLink,
)
這個版本用了手寫的 BookTagLink,主鍵是 (book_id, tag_id) 複合鍵,外加 created_at 欄位。Relationship(link_model=BookTagLink) 告訴 SQLAlchemy 用這個表做中間關聯。建立一筆「書本 + 標籤」的關聯時,只要把它們的 id 加進 join table 即可。
cascade 與約束
當作者被刪除時,他的書本該怎麼辦?答案是 cascade(cascade behavior)。SQLAlchemy 提供幾種選項:
| 選項 | 行為 | 適用情境 |
|---|---|---|
"all, delete-orphan" |
作者刪除時,書本也刪除 | 作者離開,書本沒人認領 |
"all" |
作者刪除時,書本的 author_id 設 NULL(要先允許 NULL) | 書本可以被匿名保留 |
"save-update, merge" |
預設值,不處理刪除 | 你會自己手動處理 |
設 cascade 的寫法:
# cascade 範例
class Author(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
books: list["Book"] = Relationship(
back_populates="author",
cascade_delete=True, # 作者刪除時書本也刪除(SQLite/PostgreSQL 適用)
)
cascade_delete=True 是 SQLModel 對 SQLAlchemy "all, delete-orphan" 的語法糖,會在資料庫層級加上 ON DELETE CASCADE。SQLite 與 PostgreSQL 都支援。
完整實作:作者 + 書本 + 標籤
把所有東西一次做完。我們建立一個完整的多資料表系統:
- 作者(Author):每位作者有多本書
- 書本(Book):每本書屬於一位作者、有多個標籤
- 標籤(Tag):多對多關聯到書本
# src/library_api/main.py
# 多資料表範例:作者 + 書本 + 標籤
from collections.abc import Generator
from datetime import datetime
from typing import TYPE_CHECKING, Annotated
from fastapi import Depends, FastAPI, HTTPException, Query
from pydantic import BaseModel, ConfigDict, Field
from sqlmodel import Field as SQLField, Session, SQLModel, create_engine, select
if TYPE_CHECKING:
pass
DATABASE_URL = "sqlite:///./library.db"
engine = create_engine(DATABASE_URL, echo=False)
# ---------- Join table(手動)----------
class BookTagLink(SQLModel, table=True):
book_id: int | None = Field(default=None, foreign_key="book.id", primary_key=True)
tag_id: int | None = Field(default=None, foreign_key="tag.id", primary_key=True)
created_at: datetime | None = Field(default_factory=datetime.utcnow)
# ---------- 資料表 ----------
class Author(SQLModel, table=True):
id: int | None = SQLField(default=None, primary_key=True)
name: str = SQLField(min_length=1, max_length=80, unique=True, index=True)
email: str = SQLField(unique=True, index=True)
books: list["Book"] = Relationship(
back_populates="author", cascade_delete=True
)
class Tag(SQLModel, table=True):
id: int | None = SQLField(default=None, primary_key=True)
name: str = SQLField(unique=True, index=True)
books: list["Book"] = Relationship(back_populates="tags", link_model=BookTagLink)
class Book(SQLModel, table=True):
id: int | None = SQLField(default=None, primary_key=True)
title: str = SQLField(min_length=1, max_length=200)
year: int = SQLField(ge=0, le=2100)
author_id: int | None = SQLField(default=None, foreign_key="author.id")
author: "Author | None" = Relationship(back_populates="books")
tags: list[Tag] = Relationship(back_populates="books", link_model=BookTagLink)
# 為了避免 NameError(Relationship 用字串延後解析,這裡重新綁定)
from sqlmodel import Relationship # noqa: E402
# ---------- 請求與回應模型 ----------
class TagBase(BaseModel):
name: str = Field(min_length=1, max_length=40)
class TagPublic(TagBase):
id: int
model_config = ConfigDict(from_attributes=True)
class AuthorBase(BaseModel):
name: str = Field(min_length=1, max_length=80)
email: str = Field(pattern=r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
class AuthorPublic(AuthorBase):
id: int
model_config = ConfigDict(from_attributes=True)
class BookBase(BaseModel):
title: str = Field(min_length=1, max_length=200)
year: int = Field(ge=0, le=2100)
author_id: int
class BookCreate(BookBase):
tag_ids: list[int] = Field(default_factory=list)
class BookPublic(BookBase):
id: int
author: AuthorPublic | None = None
tags: list[TagPublic] = Field(default_factory=list)
model_config = ConfigDict(from_attributes=True)
# ---------- 資料庫生命週期 ----------
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def get_session() -> Generator[Session, None, None]:
with Session(engine) as session:
yield session
SessionDep = Annotated[Session, Depends(get_session)]
app = FastAPI(title="圖書館 API(多資料表)", version="0.1.0")
@app.on_event("startup")
def on_startup():
create_db_and_tables()
# ---------- 作者端點 ----------
@app.post("/authors", response_model=AuthorPublic, status_code=201)
def create_author(payload: AuthorBase, session: SessionDep):
author = Author(**payload.model_dump())
session.add(author)
session.commit()
session.refresh(author)
return author
@app.get("/authors", response_model=list[AuthorPublic])
def list_authors(session: SessionDep):
return list(session.exec(select(Author)).all())
@app.get("/authors/{author_id}", response_model=AuthorPublic)
def get_author(session: SessionDep, author_id: int):
author = session.get(Author, author_id)
if author is None:
raise HTTPException(status_code=404, detail="author not found")
return author
# ---------- 書本端點 ----------
@app.post("/books", response_model=BookPublic, status_code=201)
def create_book(payload: BookCreate, session: SessionDep):
# 確認作者存在
author = session.get(Author, payload.author_id)
if author is None:
raise HTTPException(status_code=404, detail="author not found")
# 建立書本(先不處理 tags)
book = Book(
title=payload.title,
year=payload.year,
author_id=payload.author_id,
)
session.add(book)
session.commit()
session.refresh(book)
# 處理標籤:建立 join table 紀錄
for tag_id in payload.tag_ids:
tag = session.get(Tag, tag_id)
if tag is None:
continue
link = BookTagLink(book_id=book.id, tag_id=tag_id)
session.add(link)
session.commit()
session.refresh(book)
return book
@app.get("/books/{book_id}", response_model=BookPublic)
def get_book(session: SessionDep, book_id: int):
book = session.get(Book, book_id)
if book is None:
raise HTTPException(status_code=404, detail="book not found")
return book
@app.get("/books/{book_id}/with-relations", response_model=BookPublic)
def get_book_with_relations(session: SessionDep, book_id: int):
"""示範主動載入關聯"""
book = session.get(Book, book_id)
if book is None:
raise HTTPException(status_code=404, detail="book not found")
# 觸發 lazy load
_ = book.author
_ = book.tags
return book
# ---------- 標籤端點 ----------
@app.post("/tags", response_model=TagPublic, status_code=201)
def create_tag(payload: TagBase, session: SessionDep):
tag = Tag(**payload.model_dump())
session.add(tag)
session.commit()
session.refresh(tag)
return tag
@app.get("/tags", response_model=list[TagPublic])
def list_tags(session: SessionDep):
return list(session.exec(select(Tag)).all())
# ---------- 作者刪除(cascade 示範) ----------
@app.delete("/authors/{author_id}", status_code=204)
def delete_author(session: SessionDep, author_id: int):
author = session.get(Author, author_id)
if author is None:
raise HTTPException(status_code=404, detail="author not found")
session.delete(author)
# 因為 cascade_delete=True,作者的所有書會被一併刪除
session.commit()
return None
# 啟動:uvicorn library_api.main:app --reload --port 8000
這個版本做了幾個重要的設計選擇:
- 請求與回應模型分離:
BookCreate接收tag_ids: list[int](扁平 ID 串列),回應模型BookPublic巢狀包含Author與Tag物件,方便前端使用。 - cascade_delete=True:刪除作者時,他的書本會被一起刪除(這是合理的設計——作者離開了,沒人認領的書也應該刪掉)。
- 手動 join table:用
BookTagLink中間表來表達書本與標籤的多對多,並加上created_at欄位記錄建立時間。 - lazy loading:存取
book.author或book.tags才會去打 SQL;如果整本書本沒用到這些關聯,就不會浪費查詢。
用 httpx 走完整個生命週期:
# tests/test_relations.py
import httpx
BASE = "http://127.0.0.1:8000"
def main():
with httpx.Client(base_url=BASE, timeout=5.0) as client:
# 1. 建立作者
r = client.post(
"/authors",
json={"name": "Hao", "email": "hao@example.com"},
)
hao_id = r.json()["id"]
print(f"建立作者:{r.status_code} id={hao_id}")
# 輸出(範例):建立作者:201 id=1
# 2. 建立標籤
r1 = client.post("/tags", json={"name": "Python"})
r2 = client.post("/tags", json={"name": "Web"})
py_id = r1.json()["id"]
web_id = r2.json()["id"]
print(f"建立標籤:{r1.status_code}, {r2.status_code}")
# 輸出(範例):建立標籤:201, 201
# 3. 建立書本(帶作者 + 標籤)
r = client.post(
"/books",
json={
"title": "Python 程式設計實戰",
"year": 2024,
"author_id": hao_id,
"tag_ids": [py_id],
},
)
book_id = r.json()["id"]
print(f"建立書本:{r.status_code} id={book_id}")
# 輸出(範例):建立書本:201 id=1
# 4. 讀取書本(含關聯)
r = client.get(f"/books/{book_id}/with-relations")
body = r.json()
print(f"書本:{body['title']} ({body['year']})")
print(f" 作者:{body['author']['name']}")
print(f" 標籤:{[t['name'] for t in body['tags']]}")
# 輸出(範例):
# 書本:Python 程式設計實戰 (2024)
# 作者:Hao
# 標籤:['Python']
# 5. 建立第二本書(有兩個標籤)
r = client.post(
"/books",
json={
"title": "FastAPI 入門",
"year": 2025,
"author_id": hao_id,
"tag_ids": [py_id, web_id],
},
)
second_id = r.json()["id"]
print(f"建立第二本書:{r.status_code} id={second_id}")
# 輸出(範例):建立第二本書:201 id=2
# 6. 查作者的詳細資料(透過 /authors/{id})
r = client.get(f"/authors/{hao_id}")
print(f"作者 {r.json()['name']} 的資料:{r.json()}")
# 注意:這個回應沒有包含 books(AuthorPublic 沒列 books 欄位)
# 7. 刪除作者(cascade 刪除書本)
r = client.delete(f"/authors/{hao_id}")
print(f"刪除作者:{r.status_code}")
# 輸出(範例):刪除作者:204
# 8. 確認書本也被刪了
r = client.get(f"/books/{book_id}")
print(f"讀取書本 {book_id}:{r.status_code}")
# 輸出(範例):讀取書本 1:404
main()
這個迴圈展示了關聯設計的全部功能:建立帶關聯的書本、讀取時自動展開關聯、cascade 刪除。實際輸出會依執行順序略有不同,但結構應該一致。注意步驟 6:AuthorPublic 沒列 books 欄位,所以 GET /authors/{id} 不會包含書本串列——這是刻意設計,避免不小心一次把所有書都傳回去。
常見錯誤與踩雷
第一個常見的踩雷是「忘記 cascade_delete 導致刪不掉」。如果 Author 還有 Book,你 DELETE author 會失敗(FK constraint)。兩種解法:在 FK 設 cascade_delete=True(資料庫層級自動處理),或在程式裡先刪除所有 Book 再刪 Author(手動處理)。前者比較不容易出錯,建議優先用。
第二個是「N+1 查詢問題」。當你取 100 本書,然後對每本書都存取 book.author,SQLAlchemy 會打 100 次 SELECT 查作者——這就是 N+1。對小資料沒感覺,對大資料會明顯變慢。解法是用 selectinload 或 joinedload 一次把所有 author 載入:
# 解決 N+1:用 joinedload 一次載入
from sqlmodel import select
from sqlalchemy.orm import selectinload
statement = select(Book).options(selectinload(Book.author), selectinload(Book.tags))
books = list(session.exec(statement).all())
# 現在存取 book.author 與 book.tags 都不會額外查詢
第三個是「迴圈 import」。當 Author 引用 Book、Book 引用 Author,兩個檔案互相 import 會出錯。解法是用字串型別("Book")延後解析,或用 TYPE_CHECKING 區塊只在型別檢查時 import。本系列從現在開始所有跨檔案引用都用字串型別。
效能與實務提醒
關聯設計對效能的最大影響是 JOIN 查詢的成本。對小資料(一千筆以下)沒感覺,但對大資料(十萬筆以上)需要:為外鍵欄位加索引(Field(foreign_key=..., index=True))、謹慎使用 lazy load(用 selectinload 預載入)、避免一次 JOIN 太多表(超過 4 個表通常需要重新考慮設計)。
cascade 是雙面刃。"all, delete-orphan" 會讓刪除作者變得「自動」,但也意味著你不小心刪掉作者就會連帶刪除所有書本(資料救不回來)。實務上建議:對「必然隨父層級消滅」的子資料(例如書本的章節)用 cascade;對「應該被保留」的子資料(例如訂單的客戶即使客戶刪除也要保留訂單紀錄)用 SET NULL 或手動處理。
多對多關係在大資料量時要特別注意 join table 的大小。如果一本書有 100 個標籤、1 萬本書,join table 就有 100 萬筆。對 join table 也要加索引((book_id, tag_id) 複合主鍵會自動加索引,但若要單獨查 WHERE book_id = ? 或 WHERE tag_id = ?,建議另外加索引)。
小結
今天進入資料庫設計的核心——關聯。我們用 SQLModel 的外鍵與 Relationship 機制,把昨天的單一書本表升級成「作者 + 書本 + 標籤」三資料表,並示範了一對多、多對多、cascade 刪除等概念。整套 API 已經具備一個小型圖書館系統的雛形,後續 Day 35 的「預約管理系統」會用類似的關聯結構貫穿整個專案篇。學到這裡,你已經具備寫一個「多資料表」的後端所需的所有基礎。
進階主題:關聯查詢的效能與索引策略
今天前面展示了關聯設計的基本寫法,但實務上「跨多張表的查詢」往往是效能瓶頸所在。這裡深入討論關聯查詢的最佳化策略,讓你在資料量成長時也能保持回應速度。
# eager_loading.py
# 用 selectinload 與 joinedload 預載入,避免 N+1
from sqlalchemy.orm import joinedload, selectinload
from sqlmodel import Session, select
def get_books_with_author(session: Session, book_id: int):
"""joinedload:用 JOIN 一次拿回 book 與 author"""
statement = select(Book).where(Book.id == book_id).options(
joinedload(Book.author)
)
return session.exec(statement).one()
def list_books_with_eager_tags(session: Session, limit: int = 20):
"""selectinload:用子查詢一次拿回所有 tags"""
statement = select(Book).limit(limit).options(
selectinload(Book.tags),
selectinload(Book.author),
)
return list(session.exec(statement).all())
def count_books_per_author(session: Session):
"""聚合查詢:每個作者有幾本書"""
statement = (
select(Author, func.count(Book.id).label("book_count"))
.outerjoin(Book, Book.author_id == Author.id)
.group_by(Author.id)
)
return session.exec(statement).all()
# 範例輸出:[Author(id=1, name='Hao'), 3, Author(id=2, name='Wei'), 1]
這三個查詢展示關聯查詢的三種典型模式。joinedload 用 SQL JOIN 一次拿回資料,適合「一對一」或「一對少量」的關係(例如 Book 與 Author);selectinload 用「子查詢」分批拿回,適合「一對多」的關係(例如 Book 與多個 Tags)。兩者的差別是「資料量小用 JOIN、資料量大用子查詢」。group_by 搭配 count 是典型的聚合查詢,能回答「每位作者有幾本書」這種統計問題。
第二個進階主題是「複合索引與排序」。當你經常需要「依作者篩選 + 依年份排序」,可以建立複合索引 (author_id, year DESC),讓資料庫能直接用索引完成排序,不需要額外的步驟。
# composite_index.py
# 複合索引:對經常一起查詢的欄位建立索引
from sqlmodel import Field, Session, SQLModel, create_engine, text
class Book(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
title: str
author_id: int = Field(foreign_key="author.id")
year: int
engine = create_engine("sqlite:///./library.db")
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
# 對 (author_id, year) 建立複合索引
session.execute(
text("CREATE INDEX IF NOT EXISTS idx_author_year ON book (author_id, year DESC)")
)
session.commit()
# 之後的查詢 WHERE author_id = ? ORDER BY year DESC 就能用這個索引
複合索引的設計原則:「把等值查詢的欄位放前面、範圍查詢或排序的欄位放後面」。例如 WHERE author_id = ? ORDER BY year DESC,最適合 (author_id, year DESC) 這個順序。如果順序反過來變成 (year, author_id),索引的效用會大打折扣。理解這個原則對資料量大時的效能調校特別有幫助。
第三個進階主題是「cascade 的實際行為」。前面介紹了 cascade_delete=True 會在資料庫層級加上 ON DELETE CASCADE,但實務上你應該了解這個行為對應用程式的影響——一旦刪除父層級,子層級就「靜悄悄」地被刪掉,沒有任何警告。
# cascade_demo.py
# 展示 cascade 的實際觸發流程
from sqlmodel import Session, select
def delete_author_with_books(session: Session, author_id: int) -> int:
"""刪除作者,cascade 把他的書也刪掉;回傳被刪的書本數"""
author = session.get(Author, author_id)
if author is None:
return 0
# 先計算會被影響的書本數
book_count = session.exec(
select(func.count(Book.id)).where(Book.author_id == author_id)
).one()
# 觸發 cascade
session.delete(author)
session.commit()
# books 資料表裡所有 author_id == author_id 的紀錄會被一併刪除
return book_count
def safe_delete_with_count(session: Session, author_id: int, confirm: bool = False):
"""安全版本:先確認再刪"""
book_count = session.exec(
select(func.count(Book.id)).where(Book.author_id == author_id)
).one()
if not confirm and book_count > 0:
return {
"requires_confirmation": True,
"would_delete_books": book_count,
}
session.delete(session.get(Author, author_id))
session.commit()
return {"deleted": True, "books_affected": book_count}
這兩個版本示範「cascade 刪除」的兩種使用方式。第一個版本直接刪除,適合管理後台的批次操作;第二個版本會先回傳「會被影響的書本數」,讓呼叫端決定是否真的要刪,這對前端 UI(顯示「刪除作者會一併刪除 5 本書,確定嗎?」)特別有用。實務上建議所有破壞性操作都採用第二種「先回傳影響、再讓使用者確認」的設計,避免誤刪。
另一個 cascade 的進階議題是「多對多的 join table 要不要 cascade」。以書本與標籤為例,刪除一本書時,join table 裡的對應紀錄也應該被清掉——否則會留下「指向不存在書本」的孤兒紀錄。cascade_delete=True 在 Relationship(link_model=BookTagLink) 上會把這個連帶刪除一起處理掉。實務上建議把「join table 的 cascade」一律設成 True,避免孤兒紀錄累積。
最後一個關聯設計的提醒:正規化是手段不是目的。我們拆出「作者」表是為了解決「作者 email 重複」的問題,但如果你的系統只有「作者姓名」一個欄位、且作者永遠只有 1-2 本書,把作者拆成獨立表的成本(多一次查詢、cascade 設定複雜度)反而比「好處」大。這就是過度設計(over-engineering)。判斷的原則是:「這個欄位是否會被獨立查詢或更新?」如果答案是「否」,可以先留在原表,等真的有需求再拆。
結語
今天我們把資料庫從單表升級成可關聯的多表結構,這是後端開發的關鍵里程碑。明天,我們要進入另一個維度:遷移管理(Alembic)。當資料模型隨著需求演進時(例如加欄位、改型別、加索引),怎麼讓「正式環境的資料庫」與「程式碼」同步演進而不會出錯,是所有真實專案的必修課。學完 Alembic,你就具備「修改資料庫結構而不破壞既有資料」的能力,這也是「腳本」與「系統」之間又一條清楚的界線。
延伸資源
- SQLModel Relationship 教學(2025):
https://sqlmodel.tiangolo.com/tutorial/relationship-attributes/ - SQLAlchemy 2.0 Relationship(2024):
https://docs.sqlalchemy.org/en/20/orm/relationships.html - 資料庫正規化(Database Design,Wikipedia,2025):
https://en.wikipedia.org/wiki/Database_normalization - Use The Index, Luke:N+1 與 JOIN 最佳化(2014):
https://use-the-index-luke.com/sql/join - SQLModel cascade_delete 介紹(2025):
https://sqlmodel.tiangolo.com/tutorial/relationship-attributes/cascade-delete-relationships/
留言
張貼留言