【FastAPI 學習筆記 EP.18】關聯式資料

什麼是關聯式資料？

關聯式資料指的是在資料庫的兩個資料表（Table）之間建立連結，讓程式能透過邏輯關係輕鬆存取相關數據。例如：從「使用者 (User)」直接取得該人所有的「項目 (Items)」，而不需要使用複雜的 SQL Join。

關聯式資料基礎應用

1. ForeignKey()：在子表欄位中使用，鎖定關聯目標
2. relationship()：讓程式能像存取物件屬性般取得關聯資料
3. back_populates：父表與子表需同時定義，確保雙向都能讀取
4. lazy="selectin"：避免 N+1 Query，API 回傳關聯資料時優先使用

常見錯誤與解決方法

1. 少了 ForeignKey：未宣告外鍵導致關聯失效，補上 ForeignKey("table.column")
2. 未設定 back_populates：雙向資料不同步，兩邊 relationship 都加上對應欄位

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, ForeignKey, select
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    relationship,
    Session,
)
from pydantic import BaseModel, ConfigDict

engine = create_engine(
    "sqlite:///./test.db",
    connect_args={"check_same_thread": False},
)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    
    # 一對多關聯：一個 User 有多個 Item
    items: Mapped[list["Item"]] = relationship(
        back_populates="owner",
        lazy="selectin",  # 優化查詢效能，避免 N+1
    )

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column()
    
    # 外鍵：儲存關聯的 User ID
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    
    # 多對一關聯：多個 Item 屬於一個 User
    owner: Mapped["User"] = relationship(back_populates="items")

# 建立資料表 (只需執行一次)
Base.metadata.create_all(engine)

class ItemSchema(BaseModel):
    title: str

    model_config = ConfigDict(from_attributes=True)

class UserSchema(BaseModel):
    id: int
    name: str
    items: list[ItemSchema] = []
    
    model_config = ConfigDict(from_attributes=True)

app = FastAPI()

def get_db():
    with Session(engine) as session:
        yield session

@app.post("/seed")
def seed_data(db: Session = Depends(get_db)):
    user = User(name="Alice")
    user.items = [Item(title="Book"), Item(title="Pen")]
    
    db.add(user)
    db.commit()
    return {"status": "ok", "message": "已建立 Alice 與她的兩筆資料"}

@app.get("/users", response_model=list[UserSchema])
def read_users(db: Session = Depends(get_db)):

    stmt = select(User)
    users = db.scalars(stmt).all()
    return users