luk

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

01-4. 請求體與回應模型

⏱️ 閱讀時間： 15 分鐘 🎯 難度： ⭐⭐ (基礎)

🤔 一句話解釋

請求體是客戶端發送給 API 的資料，回應模型定義 API 返回的資料格式。

Client                              Server
  │                                    │
  │  POST /users                       │
  │  ─────────────────────────────────▶│
  │  請求體 (Request Body):            │
  │  {"name": "John", "email": "..."}  │
  │                                    │
  │  回應 (Response):                  │
  │  ◀─────────────────────────────────│
  │  {"id": 1, "name": "John", ...}    │
  │                                    │

📥 請求體 (Request Body)

基本用法

使用 Pydantic 模型定義請求體結構：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 定義請求體模型
class UserCreate(BaseModel):
    name: str
    email: str
    age: int

@app.post("/users")
async def create_user(user: UserCreate):
    # user 自動被驗證和轉換
    return {
        "message": f"使用者 {user.name} 建立成功",
        "user": user
    }

請求：

curl -X POST "http://localhost:8000/users" \
     -H "Content-Type: application/json" \
     -d '{"name": "John", "email": "john@example.com", "age": 25}'

回應：

{
    "message": "使用者 John 建立成功",
    "user": {
        "name": "John",
        "email": "john@example.com",
        "age": 25
    }
}

帶預設值的欄位

from pydantic import BaseModel
from typing import Optional

class ItemCreate(BaseModel):
    name: str                           # 必填
    description: Optional[str] = None   # 選填，預設 None
    price: float                        # 必填
    quantity: int = 1                   # 選填，預設 1
    is_active: bool = True              # 選填，預設 True

@app.post("/items")
async def create_item(item: ItemCreate):
    return item

最小請求：

{
    "name": "筆記本",
    "price": 99.9
}

完整請求：

{
    "name": "筆記本",
    "description": "A5 大小，200頁",
    "price": 99.9,
    "quantity": 10,
    "is_active": true
}

巢狀模型

from pydantic import BaseModel
from typing import Optional, List

class Address(BaseModel):
    city: str
    street: str
    zip_code: str

class Contact(BaseModel):
    phone: Optional[str] = None
    email: str

class UserCreate(BaseModel):
    name: str
    address: Address              # 巢狀模型
    contacts: List[Contact]       # 模型列表

@app.post("/users")
async def create_user(user: UserCreate):
    return user

請求：

{
    "name": "John",
    "address": {
        "city": "台北市",
        "street": "信義路100號",
        "zip_code": "110"
    },
    "contacts": [
        {"email": "john@work.com", "phone": "0912345678"},
        {"email": "john@personal.com"}
    ]
}

📤 回應模型 (Response Model)

為什麼需要回應模型？

# ❌ 不好的做法：直接回傳資料庫模型
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = database.get_user(user_id)
    return user  # 可能包含密碼等敏感資訊！

# ✅ 好的做法：使用回應模型過濾
class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    # 沒有 password！

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    user = database.get_user(user_id)
    return user  # 自動過濾，只返回指定欄位

基本用法

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import Optional

app = FastAPI()

# 請求模型（建立時用）
class UserCreate(BaseModel):
    name: str
    email: EmailStr
    password: str  # 請求時需要密碼

# 回應模型（回傳時用）
class UserResponse(BaseModel):
    id: int
    name: str
    email: EmailStr
    created_at: datetime
    # 沒有 password！不會回傳給客戶端

@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
    # 模擬建立使用者
    new_user = {
        "id": 1,
        "name": user.name,
        "email": user.email,
        "password": "hashed_password",  # 這個不會被回傳
        "created_at": datetime.now()
    }
    return new_user  # FastAPI 會根據 response_model 過濾

請求：

{
    "name": "John",
    "email": "john@example.com",
    "password": "secret123"
}

回應（密碼被過濾）：

{
    "id": 1,
    "name": "John",
    "email": "john@example.com",
    "created_at": "2025-12-17T10:30:00"
}

列表回應

from typing import List

class ItemResponse(BaseModel):
    id: int
    name: str
    price: float

@app.get("/items", response_model=List[ItemResponse])
async def list_items():
    items = [
        {"id": 1, "name": "Item 1", "price": 10.0, "secret": "xxx"},
        {"id": 2, "name": "Item 2", "price": 20.0, "secret": "yyy"},
    ]
    return items  # secret 欄位會被過濾

response_model 進階選項

from pydantic import BaseModel
from typing import Optional

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

# 排除未設定的欄位
@app.get(
    "/items/{item_id}",
    response_model=Item,
    response_model_exclude_unset=True  # 不回傳 None 的欄位
)
async def get_item(item_id: int):
    return {"name": "Item", "price": 10.0}
    # 回應: {"name": "Item", "price": 10.0}
    # 不會包含 description 和 tax

# 排除特定欄位
@app.get(
    "/items-no-tax/{item_id}",
    response_model=Item,
    response_model_exclude={"tax"}  # 排除 tax
)
async def get_item_no_tax(item_id: int):
    return {"name": "Item", "price": 10.0, "tax": 1.0}
    # 回應不會包含 tax

# 只包含特定欄位
@app.get(
    "/items-name/{item_id}",
    response_model=Item,
    response_model_include={"name", "price"}  # 只包含這些
)
async def get_item_name(item_id: int):
    return {"name": "Item", "price": 10.0, "tax": 1.0}

🎨 請求體 + 路徑參數 + 查詢參數

可以同時使用多種參數來源：

from fastapi import FastAPI, Path, Query
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class ItemUpdate(BaseModel):
    name: Optional[str] = None
    price: Optional[float] = None
    description: Optional[str] = None

@app.put("/items/{item_id}")
async def update_item(
    # 路徑參數
    item_id: int = Path(..., ge=1, description="項目 ID"),

    # 查詢參數
    notify: bool = Query(False, description="是否發送通知"),

    # 請求體
    item: ItemUpdate = None
):
    result = {
        "item_id": item_id,
        "notify": notify,
        "updated_fields": item.model_dump(exclude_unset=True) if item else {}
    }
    return result

請求：

curl -X PUT "http://localhost:8000/items/123?notify=true" \
     -H "Content-Type: application/json" \
     -d '{"name": "新名稱", "price": 199.99}'

回應：

{
    "item_id": 123,
    "notify": true,
    "updated_fields": {
        "name": "新名稱",
        "price": 199.99
    }
}

🔄 多個請求體

有時候需要接收多個不同的請求體：

from fastapi import FastAPI, Body
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

class User(BaseModel):
    username: str

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item,
    user: User,
    importance: int = Body(..., ge=1, le=5)  # 單一值也可以放在 body
):
    return {
        "item_id": item_id,
        "item": item,
        "user": user,
        "importance": importance
    }

請求：

{
    "item": {
        "name": "Foo",
        "price": 50.5
    },
    "user": {
        "username": "john"
    },
    "importance": 5
}

使用 Body(embed=True)

class Item(BaseModel):
    name: str
    price: float

# 不使用 embed
@app.post("/items")
async def create_item(item: Item):
    pass
# 請求體: {"name": "Foo", "price": 50.5}

# 使用 embed
@app.post("/items-embedded")
async def create_item_embedded(item: Item = Body(embed=True)):
    pass
# 請求體: {"item": {"name": "Foo", "price": 50.5}}

📝 實戰範例：部落格文章 API

from fastapi import FastAPI, Path, Query, HTTPException
from pydantic import BaseModel, Field, ConfigDict
from typing import Optional, List
from datetime import datetime
from enum import Enum

app = FastAPI(title="部落格 API")

# ===== Enums =====
class PostStatus(str, Enum):
    draft = "draft"
    published = "published"
    archived = "archived"

# ===== 請求模型 =====
class TagCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=30)

class PostCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    content: str = Field(..., min_length=10)
    summary: Optional[str] = Field(None, max_length=500)
    tags: List[str] = Field(default_factory=list)
    status: PostStatus = PostStatus.draft

class PostUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    content: Optional[str] = Field(None, min_length=10)
    summary: Optional[str] = Field(None, max_length=500)
    tags: Optional[List[str]] = None
    status: Optional[PostStatus] = None

# ===== 回應模型 =====
class AuthorResponse(BaseModel):
    id: int
    name: str
    avatar_url: Optional[str] = None

class PostResponse(BaseModel):
    id: int
    title: str
    content: str
    summary: Optional[str]
    tags: List[str]
    status: PostStatus
    author: AuthorResponse
    view_count: int
    created_at: datetime
    updated_at: Optional[datetime]

    model_config = ConfigDict(from_attributes=True)

class PostListResponse(BaseModel):
    """文章列表回應（不含完整內容）"""
    id: int
    title: str
    summary: Optional[str]
    tags: List[str]
    status: PostStatus
    author: AuthorResponse
    view_count: int
    created_at: datetime

class PaginatedResponse(BaseModel):
    """分頁回應"""
    total: int
    page: int
    per_page: int
    pages: int
    items: List[PostListResponse]

# ===== 模擬資料 =====
fake_author = AuthorResponse(id=1, name="John Doe", avatar_url=None)
fake_posts = {}
post_counter = 0

# ===== API 端點 =====
@app.get("/posts", response_model=PaginatedResponse)
async def list_posts(
    page: int = Query(1, ge=1),
    per_page: int = Query(10, ge=1, le=100),
    status: Optional[PostStatus] = None,
    tag: Optional[str] = None,
    search: Optional[str] = Query(None, min_length=2)
):
    """
    獲取文章列表

    - 支援分頁
    - 支援狀態過濾
    - 支援標籤過濾
    - 支援關鍵字搜尋
    """
    # 過濾邏輯（簡化版）
    filtered_posts = list(fake_posts.values())

    if status:
        filtered_posts = [p for p in filtered_posts if p["status"] == status]
    if tag:
        filtered_posts = [p for p in filtered_posts if tag in p["tags"]]
    if search:
        filtered_posts = [p for p in filtered_posts if search.lower() in p["title"].lower()]

    total = len(filtered_posts)
    start = (page - 1) * per_page
    end = start + per_page
    paginated = filtered_posts[start:end]

    return PaginatedResponse(
        total=total,
        page=page,
        per_page=per_page,
        pages=(total + per_page - 1) // per_page,
        items=[PostListResponse(**p, author=fake_author) for p in paginated]
    )

@app.post("/posts", response_model=PostResponse, status_code=201)
async def create_post(post: PostCreate):
    """建立新文章"""
    global post_counter
    post_counter += 1

    new_post = {
        "id": post_counter,
        "title": post.title,
        "content": post.content,
        "summary": post.summary,
        "tags": post.tags,
        "status": post.status,
        "view_count": 0,
        "created_at": datetime.now(),
        "updated_at": None
    }
    fake_posts[post_counter] = new_post

    return PostResponse(**new_post, author=fake_author)

@app.get("/posts/{post_id}", response_model=PostResponse)
async def get_post(
    post_id: int = Path(..., ge=1)
):
    """獲取單一文章"""
    if post_id not in fake_posts:
        raise HTTPException(status_code=404, detail="文章不存在")

    post = fake_posts[post_id]
    post["view_count"] += 1  # 增加瀏覽次數

    return PostResponse(**post, author=fake_author)

@app.patch("/posts/{post_id}", response_model=PostResponse)
async def update_post(
    post_id: int = Path(..., ge=1),
    post_update: PostUpdate = None
):
    """更新文章（部分更新）"""
    if post_id not in fake_posts:
        raise HTTPException(status_code=404, detail="文章不存在")

    post = fake_posts[post_id]

    if post_update:
        update_data = post_update.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            post[key] = value
        post["updated_at"] = datetime.now()

    return PostResponse(**post, author=fake_author)

@app.delete("/posts/{post_id}", status_code=204)
async def delete_post(
    post_id: int = Path(..., ge=1)
):
    """刪除文章"""
    if post_id not in fake_posts:
        raise HTTPException(status_code=404, detail="文章不存在")

    del fake_posts[post_id]

✅ 重點總結

請求體 vs 回應模型

特性	請求體 (Request Body)	回應模型 (Response Model)
用途	接收客戶端資料	過濾回傳資料
定義方式	參數型別為 Pydantic 模型	`response_model=`
驗證	自動驗證輸入	自動過濾輸出
常見場景	POST/PUT/PATCH	所有方法

常見模式

# 1. 建立用模型（接收所有資料）
class UserCreate(BaseModel):
    name: str
    email: str
    password: str  # 敏感資料

# 2. 回應用模型（過濾敏感資料）
class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    # 沒有 password

# 3. 更新用模型（全部選填）
class UserUpdate(BaseModel):
    name: Optional[str] = None
    email: Optional[str] = None
    password: Optional[str] = None

最佳實踐

分離請求和回應模型 - 不要用同一個模型
回應模型不包含敏感資料 - 密碼、token 等
使用 response_model - 確保回應格式一致
更新用模型欄位設為選填 - 支援部分更新

🎤 面試這樣答

Q: FastAPI 中如何確保 API 不會回傳敏感資料？

答案：

使用 response_model 參數指定回應模型，FastAPI 會自動過濾掉不在模型中的欄位。
class UserResponse(BaseModel):
    id: int
    name: str
    # 沒有 password

@app.get("/users/{id}", response_model=UserResponse)
async def get_user(id: int):
    user = db.get_user(id)  # 可能包含 password
    return user  # 自動過濾，不會回傳 password
這樣即使資料庫模型包含敏感欄位，回應中也不會出現。

🤓 小測驗

如何定義一個請求體？
response_model 的作用是什麼？
如何讓回應不包含值為 None 的欄位？

上一篇： 01-3. 路徑參數與查詢參數 下一篇： 01-5. 狀態碼與錯誤處理

最後更新：2025-12-17