luk

0001-01-01 約 1400 字預計閱讀 7 分鐘

03-2. 關聯關係

⏱️ 閱讀時間： 20 分鐘 🎯 難度： ⭐⭐⭐ (進階)

🤔 一句話解釋

SQLAlchemy 的 relationship 讓你用 Python 物件存取關聯資料，不用手動寫 JOIN。

🔗 關聯類型總覽

┌─────────────────────────────────────────────────────────────┐
│                      關聯類型                                │
├─────────────────────────────────────────────────────────────┤
│  One-to-Many (一對多)                                        │
│  ┌─────┐         ┌─────┐                                    │
│  │User │ ──1:N──▶│Post │  一個使用者有多篇文章               │
│  └─────┘         └─────┘                                    │
├─────────────────────────────────────────────────────────────┤
│  Many-to-One (多對一)                                        │
│  ┌─────┐         ┌────────┐                                 │
│  │Post │ ──N:1──▶│Category│  多篇文章屬於一個分類            │
│  └─────┘         └────────┘                                 │
├─────────────────────────────────────────────────────────────┤
│  One-to-One (一對一)                                         │
│  ┌─────┐         ┌───────┐                                  │
│  │User │ ──1:1──▶│Profile│  一個使用者有一個個人資料         │
│  └─────┘         └───────┘                                  │
├─────────────────────────────────────────────────────────────┤
│  Many-to-Many (多對多)                                       │
│  ┌─────┐         ┌─────┐                                    │
│  │Post │ ──N:N──▶│ Tag │  文章和標籤的多對多關係             │
│  └─────┘         └─────┘                                    │
└─────────────────────────────────────────────────────────────┘

🔢 One-to-Many（一對多）

範例：使用者與文章

from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime
from sqlalchemy.orm import relationship, Mapped, mapped_column
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
from typing import List, Optional

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(String(100), unique=True)

    # 關聯：一個使用者有多篇文章
    posts: Mapped[List["Post"]] = relationship(
        "Post",
        back_populates="author",
        cascade="all, delete-orphan"  # 刪除使用者時，也刪除他的文章
    )

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

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

    # 外鍵
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    # 關聯：文章屬於一個作者
    author: Mapped["User"] = relationship("User", back_populates="posts")

    def __repr__(self):
        return f"<Post(id={self.id}, title='{self.title}')>"

使用方式

from sqlalchemy.orm import Session

def demo_one_to_many(db: Session):
    # 建立使用者和文章
    user = User(username="john", email="john@example.com")

    # 方法 1: 從使用者端新增文章
    user.posts = [
        Post(title="First Post", content="Hello World"),
        Post(title="Second Post", content="Another post")
    ]
    db.add(user)
    db.commit()

    # 方法 2: 從文章端關聯使用者
    post = Post(
        title="Third Post",
        content="Content here",
        author=user  # 直接指定 author
    )
    db.add(post)
    db.commit()

    # 查詢：取得使用者的所有文章
    user = db.query(User).filter(User.username == "john").first()
    for post in user.posts:
        print(f"- {post.title}")

    # 查詢：取得文章的作者
    post = db.query(Post).first()
    print(f"Author: {post.author.username}")

1️⃣ One-to-One（一對一）

範例：使用者與個人資料

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)

    # 一對一關聯
    profile: Mapped[Optional["Profile"]] = relationship(
        "Profile",
        back_populates="user",
        uselist=False,  # 關鍵：設為 False 表示一對一
        cascade="all, delete-orphan"
    )

class Profile(Base):
    __tablename__ = "profiles"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    bio: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    avatar_url: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
    website: Mapped[Optional[str]] = mapped_column(String(200), nullable=True)

    # 外鍵（unique=True 確保一對一）
    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id"),
        unique=True  # 確保每個使用者只有一個 Profile
    )

    # 關聯
    user: Mapped["User"] = relationship("User", back_populates="profile")

使用方式

def demo_one_to_one(db: Session):
    # 建立使用者和個人資料
    user = User(
        username="john",
        profile=Profile(
            bio="Python developer",
            website="https://john.dev"
        )
    )
    db.add(user)
    db.commit()

    # 存取
    print(user.profile.bio)  # Python developer
    print(user.profile.user.username)  # john

🔄 Many-to-Many（多對多）

範例：文章與標籤

from sqlalchemy import Table, Column, Integer, ForeignKey

# 關聯表（不需要 Model 類別）
post_tags = Table(
    "post_tags",
    Base.metadata,
    Column("post_id", Integer, ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True)
)

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(Text)

    # 多對多關聯
    tags: Mapped[List["Tag"]] = relationship(
        "Tag",
        secondary=post_tags,  # 指定關聯表
        back_populates="posts"
    )

class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)

    # 多對多關聯
    posts: Mapped[List["Post"]] = relationship(
        "Post",
        secondary=post_tags,
        back_populates="tags"
    )

使用方式

def demo_many_to_many(db: Session):
    # 建立標籤
    python_tag = Tag(name="Python")
    fastapi_tag = Tag(name="FastAPI")
    db.add_all([python_tag, fastapi_tag])
    db.commit()

    # 建立文章並關聯標籤
    post = Post(
        title="FastAPI Tutorial",
        content="Learn FastAPI...",
        tags=[python_tag, fastapi_tag]
    )
    db.add(post)
    db.commit()

    # 查詢：取得文章的所有標籤
    post = db.query(Post).first()
    for tag in post.tags:
        print(f"- {tag.name}")

    # 查詢：取得標籤下的所有文章
    tag = db.query(Tag).filter(Tag.name == "Python").first()
    for post in tag.posts:
        print(f"- {post.title}")

    # 新增標籤到文章
    new_tag = db.query(Tag).filter(Tag.name == "FastAPI").first()
    post.tags.append(new_tag)
    db.commit()

    # 移除標籤
    post.tags.remove(new_tag)
    db.commit()

帶額外資料的多對多

# 如果關聯表需要額外欄位，就需要用 Model 類別

class PostTag(Base):
    """文章標籤關聯（帶額外資料）"""
    __tablename__ = "post_tags"

    post_id: Mapped[int] = mapped_column(
        ForeignKey("posts.id"),
        primary_key=True
    )
    tag_id: Mapped[int] = mapped_column(
        ForeignKey("tags.id"),
        primary_key=True
    )

    # 額外欄位
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    created_by: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"))

    # 關聯
    post: Mapped["Post"] = relationship("Post", back_populates="post_tags")
    tag: Mapped["Tag"] = relationship("Tag", back_populates="post_tags")

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))

    # 透過關聯 Model
    post_tags: Mapped[List["PostTag"]] = relationship(
        "PostTag",
        back_populates="post",
        cascade="all, delete-orphan"
    )

    # 方便存取（使用 association_proxy）
    @property
    def tags(self) -> List["Tag"]:
        return [pt.tag for pt in self.post_tags]

class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)

    post_tags: Mapped[List["PostTag"]] = relationship(
        "PostTag",
        back_populates="tag"
    )

📥 載入策略（Loading Strategies）

問題：N+1 查詢

# ❌ N+1 問題：會執行 1 + N 次查詢
users = db.query(User).all()
for user in users:
    print(user.posts)  # 每次存取都會查詢一次

解決方案

1. Eager Loading（積極載入）

from sqlalchemy.orm import joinedload, selectinload, subqueryload

# joinedload: 使用 JOIN（適合一對一、多對一）
stmt = select(User).options(joinedload(User.profile))
users = db.scalars(stmt).all()

# selectinload: 使用 IN 子查詢（適合一對多）
stmt = select(User).options(selectinload(User.posts))
users = db.scalars(stmt).unique().all()

# subqueryload: 使用子查詢
stmt = select(User).options(subqueryload(User.posts))
users = db.scalars(stmt).all()

2. 巢狀載入

# 載入多層關聯
stmt = select(User).options(
    selectinload(User.posts).selectinload(Post.tags)
)
users = db.scalars(stmt).unique().all()

# 現在可以存取所有資料，不會額外查詢
for user in users:
    for post in user.posts:
        for tag in post.tags:
            print(f"{user.username} - {post.title} - {tag.name}")

3. Lazy Loading 選項

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)

    # lazy 選項
    posts: Mapped[List["Post"]] = relationship(
        "Post",
        back_populates="author",
        lazy="select"  # 預設：存取時才查詢
        # lazy="joined"   # 使用 JOIN 立即載入
        # lazy="selectin" # 使用 IN 查詢立即載入
        # lazy="subquery" # 使用子查詢立即載入
        # lazy="dynamic"  # 返回 Query 物件，可以進一步過濾
    )

載入策略比較

策略	查詢數	適用場景
`select`（預設）	1 + N	不確定是否需要關聯資料
`joined`	1	一對一、多對一
`selectin`	2	一對多（推薦）
`subquery`	2	一對多（複雜過濾）

🔄 Cascade 選項

常用 Cascade

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)

    posts: Mapped[List["Post"]] = relationship(
        "Post",
        back_populates="author",
        cascade="all, delete-orphan"
        # cascade 選項：
        # - "save-update": 儲存父物件時，也儲存子物件
        # - "merge": 合併父物件時，也合併子物件
        # - "expunge": 從 Session 移除父物件時，也移除子物件
        # - "delete": 刪除父物件時，也刪除子物件
        # - "delete-orphan": 從關聯移除的子物件也會被刪除
        # - "all": 包含以上所有（除了 delete-orphan）
    )

Cascade 範例

def demo_cascade(db: Session):
    # 建立使用者和文章
    user = User(username="john")
    user.posts = [
        Post(title="Post 1", content="..."),
        Post(title="Post 2", content="...")
    ]
    db.add(user)
    db.commit()

    # 刪除使用者時，文章也會被刪除（cascade="delete"）
    db.delete(user)
    db.commit()

    # 從關聯移除文章時，文章會被刪除（cascade="delete-orphan"）
    user.posts.remove(user.posts[0])
    db.commit()  # 被移除的文章也從資料庫刪除

📝 實戰範例：部落格系統

from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, Boolean, Table
from sqlalchemy.orm import relationship, Mapped, mapped_column
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
from typing import List, Optional

Base = declarative_base()

# 多對多關聯表
post_tags = Table(
    "post_tags",
    Base.metadata,
    Column("post_id", Integer, ForeignKey("posts.id", ondelete="CASCADE"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True)
)

class User(Base):
    """使用者"""
    __tablename__ = "users"

    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))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

    # 關聯
    profile: Mapped[Optional["Profile"]] = relationship(
        "Profile",
        back_populates="user",
        uselist=False,
        cascade="all, delete-orphan"
    )
    posts: Mapped[List["Post"]] = relationship(
        "Post",
        back_populates="author",
        cascade="all, delete-orphan"
    )
    comments: Mapped[List["Comment"]] = relationship(
        "Comment",
        back_populates="author",
        cascade="all, delete-orphan"
    )

class Profile(Base):
    """使用者個人資料"""
    __tablename__ = "profiles"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    bio: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    avatar_url: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
    website: Mapped[Optional[str]] = mapped_column(String(200), nullable=True)

    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), unique=True)
    user: Mapped["User"] = relationship("User", back_populates="profile")

class Category(Base):
    """文章分類"""
    __tablename__ = "categories"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)
    slug: Mapped[str] = mapped_column(String(50), unique=True)
    description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)

    posts: Mapped[List["Post"]] = relationship("Post", back_populates="category")

class Tag(Base):
    """標籤"""
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)
    slug: Mapped[str] = mapped_column(String(50), unique=True)

    posts: Mapped[List["Post"]] = relationship(
        "Post",
        secondary=post_tags,
        back_populates="tags"
    )

class Post(Base):
    """文章"""
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(200))
    slug: Mapped[str] = mapped_column(String(200), unique=True, index=True)
    content: Mapped[str] = mapped_column(Text)
    excerpt: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    is_published: Mapped[bool] = mapped_column(default=False)
    view_count: Mapped[int] = mapped_column(default=0)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    updated_at: Mapped[Optional[datetime]] = mapped_column(onupdate=datetime.utcnow)
    published_at: Mapped[Optional[datetime]] = mapped_column(nullable=True)

    # 外鍵
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    category_id: Mapped[Optional[int]] = mapped_column(
        ForeignKey("categories.id"),
        nullable=True
    )

    # 關聯
    author: Mapped["User"] = relationship("User", back_populates="posts")
    category: Mapped[Optional["Category"]] = relationship(
        "Category",
        back_populates="posts"
    )
    tags: Mapped[List["Tag"]] = relationship(
        "Tag",
        secondary=post_tags,
        back_populates="posts"
    )
    comments: Mapped[List["Comment"]] = relationship(
        "Comment",
        back_populates="post",
        cascade="all, delete-orphan"
    )

class Comment(Base):
    """留言"""
    __tablename__ = "comments"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    content: Mapped[str] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

    # 外鍵
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    post_id: Mapped[int] = mapped_column(ForeignKey("posts.id"))
    parent_id: Mapped[Optional[int]] = mapped_column(
        ForeignKey("comments.id"),
        nullable=True
    )

    # 關聯
    author: Mapped["User"] = relationship("User", back_populates="comments")
    post: Mapped["Post"] = relationship("Post", back_populates="comments")

    # 自引用關聯（巢狀留言）
    replies: Mapped[List["Comment"]] = relationship(
        "Comment",
        backref="parent",
        remote_side=[id],
        cascade="all, delete-orphan"
    )

查詢範例

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

def get_posts_with_details(db: Session, skip: int = 0, limit: int = 10):
    """取得文章（含作者、分類、標籤）"""
    stmt = (
        select(Post)
        .where(Post.is_published == True)
        .options(
            joinedload(Post.author),
            joinedload(Post.category),
            selectinload(Post.tags)
        )
        .order_by(Post.published_at.desc())
        .offset(skip)
        .limit(limit)
    )
    return db.scalars(stmt).unique().all()

def get_post_with_comments(db: Session, slug: str):
    """取得文章和留言"""
    stmt = (
        select(Post)
        .where(Post.slug == slug)
        .options(
            joinedload(Post.author).joinedload(User.profile),
            selectinload(Post.comments).joinedload(Comment.author),
            selectinload(Post.tags)
        )
    )
    return db.scalar(stmt)

✅ 重點總結

關聯設定

關聯類型	外鍵位置	relationship 設定
一對多	多的那方	`uselist=True`（預設）
一對一	子方	`uselist=False`
多對多	關聯表	`secondary=table`

載入策略

策略	方法	適用場景
`joinedload`	JOIN	一對一、多對一
`selectinload`	IN 查詢	一對多（推薦）
`subqueryload`	子查詢	複雜過濾

🎤 面試這樣答

Q: 什麼是 N+1 問題？如何解決？

答案：

N+1 問題是指查詢 N 筆資料時，存取關聯會額外執行 N 次查詢：

users = db.query(User).all()  # 1 次查詢
for user in users:
    print(user.posts)  # N 次查詢

解決方法是使用 Eager Loading：

from sqlalchemy.orm import selectinload

stmt = select(User).options(selectinload(User.posts))
users = db.scalars(stmt).all()  # 只有 2 次查詢

上一篇： 03-1. SQLAlchemy 基礎 下一篇： 03-3. 非同步資料庫操作

最後更新：2025-12-17