API 設計：REST / GraphQL / gRPC

REST 原則、資源建模與 URL 命名 基礎 高頻

REST（Representational State Transfer）是 Roy Fielding 在 2000 年博士論文中提出的架構風格，不是協定也不是標準。它描述的是「整個 Web 為什麼能擴展到全球規模」的一組約束（constraints）。實務上大家說的「RESTful API」通常只遵守其中一部分，面試時要能分清楚「Fielding 的 REST」與「業界常見的 HTTP JSON API」。

六大約束

Client–Server

關注點分離：UI 與資料儲存分開演進。前端換框架不影響後端，後端換 DB 不影響前端。

Stateless（無狀態）

每個請求都要攜帶伺服器處理它所需的全部資訊（例如 token），伺服器不在記憶體保存「這個 client 上一步做了什麼」。好處是任何一台機器都能處理任何請求，水平擴展與故障轉移變簡單；代價是每個請求比較肥（重複帶認證資訊）。注意：「無狀態」指的是對話狀態（session state），資源狀態當然存在 DB 裡。

Cacheable

回應必須明確標示能否快取（Cache-Control、ETag），讓 client、CDN、proxy 能重用回應，減少延遲與負載。

Uniform Interface（統一介面）

REST 最核心的約束，包含四點：(1) 用 URI 識別資源；(2) 透過表述（representation，例如 JSON）操作資源；(3) 自描述訊息（method、status code、Content-Type 本身就說明了語意）；(4) HATEOAS。

Layered System

Client 不需要知道中間有沒有 LB、CDN、API Gateway。這讓我們可以隨時插入快取層、安全層。

Code on Demand（選用）

伺服器可以下發可執行程式碼（例如 JavaScript）。唯一一個選用的約束。

HATEOAS 簡述

Hypermedia As The Engine Of Application State：回應中附上「接下來可以做什麼」的連結，client 只需要知道入口 URL，就能順著連結探索整個 API，就像人類瀏覽網頁一樣。

{
  "id": "cmp_123",
  "status": "paused",
  "_links": {
    "self":   { "href": "/v1/campaigns/cmp_123" },
    "resume": { "href": "/v1/campaigns/cmp_123:resume", "method": "POST" },
    "reports":{ "href": "/v1/campaigns/cmp_123/reports" }
  }
}

實務上 HATEOAS 很少被完整採用：大部分 client 是工程師照文件寫死的，動態探索的好處不大，且增加回應體積。比較常見的「半套」是分頁回應附 next 連結、建立資源後用 Location header 指向新資源。

Richardson 成熟度模型

URL 命名規範與資源建模 基礎 高頻

URL 的設計直接反映你對領域模型的理解。原則是：URL 是名詞（資源），動作交給 HTTP method。

非 CRUD 的動作怎麼辦？

真實業務一定有「暫停廣告活動」「重送訊息」「取消任務」這種不屬於 CRUD 的動作。常見三種做法：

1. 建模成狀態欄位的更新：PATCH /campaigns/123 body {"status":"paused"}。最 REST，但狀態轉移有副作用（例如通知、計費）時語意不夠明確。
2. 建模成子資源：POST /campaigns/123/pauses 或 POST /jobs/9/cancellation。把「動作」名詞化，可以留下歷史紀錄（誰在何時暫停）。
3. Custom method（Google API Design Guide 風格）：POST /campaigns/123:pause。用冒號區隔，清楚表達這是動作，Google Cloud API 大量使用。

JSON 欄位慣例

• 欄位命名選 snake_case 或 camelCase 一種並全 API 一致（Python 後端常用 snake_case；JS 前端偏好 camelCase）。
• 時間一律用 ISO 8601 / RFC 3339 並帶時區："2026-11-03T10:00:00Z"。不要用本地時間字串。
• 金額用整數最小單位（"amount_micros": 1500000）或字串十進位，避免浮點誤差。廣告產業常用 micros（百萬分之一）表示出價。
• 大整數 ID（超過 2⁵³）用字串，避免 JavaScript Number 精度丟失。
• 列舉值用字串（"status": "active"）而不是數字，可讀且易擴充。
• 頂層回傳物件而非陣列（{"data": [...]}），方便日後加 next_cursor 等欄位而不破壞相容性。

HTTP 語意：Method、Status Code 與錯誤格式 基礎 高頻

兩個關鍵詞：

Safe（安全）

不改變伺服器資源狀態（唯讀）。爬蟲、預取、快取可以放心呼叫。

Idempotent（冪等）

同樣的請求執行 1 次與 N 次，伺服器端狀態相同。注意是狀態相同，不是回應相同：第一次 DELETE 回 204，第二次回 404，仍然是冪等的。

PUT vs POST vs PATCH

• POST /campaigns：由伺服器決定新資源 ID，重送會建立兩筆 → 不冪等。
• PUT /campaigns/cmp_123：client 指定 ID 並給完整內容；漏給的欄位應被視為清空/預設值。重送結果相同 → 冪等。
• PATCH /campaigns/cmp_123：只送要改的欄位。{"budget": 100} 是冪等的；但 {"op":"increment","path":"/budget","value":10} 這類相對操作就不是。所以 HTTP 規範說 PATCH「不保證」冪等。

Status Code 的正確使用 基礎 高頻

Status code 是給「機器」看的：client library、LB、監控、重試邏輯都依賴它。永遠回 200 然後在 body 寫 error 是最常見的反模式，它會讓監控的錯誤率失真、讓 CDN 快取錯誤回應、讓重試邏輯失效。

哪些錯誤可以重試？

一般原則：4xx 是 client 的問題，原封不動重試沒用（例外：408、409 處理中、425、429）；5xx 是 server 的問題，可以退避重試（但 501 Not Implemented 不用）。非冪等請求即使遇到 5xx/timeout，也必須帶 Idempotency-Key 才能重試。

錯誤回應格式：RFC 9457 Problem Details 進階

錯誤格式要一致、機器可解析、人類可讀。RFC 9457（2023 年取代 RFC 7807）定義了 application/problem+json：

HTTP/1.1 422 Unprocessable Content
Content-Type: application/problem+json

{
  "type": "https://api.example.com/problems/validation-error",
  "title": "Request validation failed",
  "status": 422,
  "detail": "2 fields failed validation",
  "instance": "/v1/messages/req_8f2c",
  "errors": [
    { "field": "to", "code": "invalid_line_user_id", "message": "must start with U and be 33 chars" },
    { "field": "messages", "code": "too_many_items", "message": "at most 5 messages per request" }
  ],
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736"
}

type

錯誤類型的 URI（可以是文件連結），client 用它做程式判斷。預設 about:blank。

title

簡短、人類可讀、同一 type 固定不變的摘要。

status

HTTP status code 的副本（方便 log 時 body 與 header 分離）。

detail

針對這次發生的具體說明。

instance

這次錯誤發生的識別 URI。

擴充欄位

可自由加，例如 errors（欄位級錯誤）、trace_id（讓客服/工程師對應 log 與 trace）、retry_after。

分頁、過濾、排序與 Sparse Fieldsets 進階 高頻

Offset 分頁

GET /campaigns?limit=50&offset=1000 → SQL LIMIT 50 OFFSET 1000。

• 優點：實作簡單、可跳頁（第 37 頁）、可算總頁數。
• 缺點 1：深分頁效能差——DB 必須掃過並丟棄前 offset 筆，OFFSET 1000000 需要讀一百萬筆，複雜度 O(offset + limit)。
• 缺點 2：資料漂移——翻頁之間有新資料插入，下一頁會出現重複；有資料刪除則會漏掉。
• 缺點 3：COUNT(*) 在大表上很貴（InnoDB 沒有快取精確列數，需要掃描索引）。

Cursor / Keyset 分頁

用「上一頁最後一筆的排序鍵」當起點：

-- 第一頁
SELECT id, name, created_at FROM campaigns
WHERE advertiser_id = 42
ORDER BY created_at DESC, id DESC
LIMIT 51;            -- 多取 1 筆判斷 has_more

-- 下一頁：cursor 解碼出 (last_created_at, last_id)
SELECT id, name, created_at FROM campaigns
WHERE advertiser_id = 42
  AND (created_at, id) < ('2026-11-01 10:00:00', 98765)   -- row value comparison
ORDER BY created_at DESC, id DESC
LIMIT 51;
-- 需要複合索引 (advertiser_id, created_at, id)

• 排序鍵必須唯一且穩定：只用 created_at 會在同一秒多筆時漏資料，所以加上 id 當 tie-breaker。
• 有索引時每頁成本 O(log N + limit)，與第幾頁無關。
• 插入/刪除不會造成重複或遺漏。
• 缺點：不能跳頁、難以顯示總頁數；排序欄位改變時 cursor 需重新設計。
• MySQL 對 row value comparison (a, b) < (x, y) 的索引利用在舊版較差，必要時展開成 a < x OR (a = x AND b < y)。

Cursor 應該是不透明的（base64 編碼的 JSON，必要時加 HMAC 簽章防竄改），讓伺服器保有日後改變實作的自由：

import base64, json, hmac, hashlib

SECRET = b"rotate-me"

def encode_cursor(created_at: str, id_: int) -> str:
    raw = json.dumps({"c": created_at, "i": id_}, separators=(",", ":")).encode()
    sig = hmac.new(SECRET, raw, hashlib.sha256).digest()[:8]
    return base64.urlsafe_b64encode(sig + raw).decode().rstrip("=")

def decode_cursor(cursor: str) -> tuple[str, int]:
    padded = cursor + "=" * (-len(cursor) % 4)
    blob = base64.urlsafe_b64decode(padded)
    sig, raw = blob[:8], blob[8:]
    expected = hmac.new(SECRET, raw, hashlib.sha256).digest()[:8]
    if not hmac.compare_digest(sig, expected):
        raise ValueError("invalid cursor")
    d = json.loads(raw)
    return d["c"], d["i"]

回傳格式

{
  "data": [ { "id": "cmp_9", "name": "Black Friday" } ],
  "pagination": {
    "next_cursor": "q1Zf0mJ9eyJjIjoiMjAyNi0xMS0wMVQxMDowMDowMFoiLCJpIjo5ODc2NX0",
    "has_more": true
  },
  "links": { "next": "/v1/campaigns?limit=50&cursor=q1Zf0mJ9..." }
}

也可以用 RFC 8288 的 Link header：Link: </v1/campaigns?cursor=abc>; rel="next"，GitHub API 就是這樣做。

過濾、排序、sparse fieldsets

GET /v1/campaigns?status=active,paused&created_at[gte]=2026-10-01&sort=-spend,name&fields=id,name,spend&limit=50

• 過濾：簡單等值用 ?status=active；範圍用 created_at[gte]= 或 created_after=；多值用逗號。只開放有索引支撐的欄位，否則任何人都能打出全表掃描。
• 排序：sort=-spend,name（減號表示降冪，JSON:API 慣例）。用白名單驗證欄位，絕不能直接拼進 SQL（SQL injection）。
• Sparse fieldsets：fields=id,name 只回需要的欄位，減少頻寬與序列化成本——這其實是 GraphQL 想解決的 over-fetching 問題的 REST 版解法。
• Embedding：?expand=advertiser（Stripe 風格）一次帶回關聯資源，解決 under-fetching / 多次往返。

版本控制、OpenAPI 與 API-first 進階

什麼是 breaking change？

配套措施：client 必須遵守 Tolerant Reader（忽略不認識的欄位）；舊版下線前用 Deprecation 與 Sunset header（RFC 8594）預告；監控每個版本的流量，確定歸零才關。

OpenAPI 與 API-first 開發 基礎

OpenAPI Specification（前身 Swagger）是描述 HTTP API 的機器可讀格式（YAML/JSON），目前主流為 3.0 / 3.1（3.1 與 JSON Schema 2020-12 完全相容）。Swagger 則是周邊工具品牌（Swagger UI、Swagger Editor、Codegen）。

openapi: 3.1.0
info:
  title: Push Messaging API
  version: 1.0.0
paths:
  /v1/messages:
    post:
      operationId: createMessage
      security: [{ bearerAuth: [messages:send] }]
      parameters:
        - in: header
          name: Idempotency-Key
          required: true
          schema: { type: string, maxLength: 64 }
      requestBody:
        required: true
        content:
          application/json:
            schema: { $ref: '#/components/schemas/MessageCreate' }
      responses:
        '202':
          description: Accepted
          content:
            application/json:
              schema: { $ref: '#/components/schemas/Message' }
        '422': { $ref: '#/components/responses/Problem' }
        '429': { $ref: '#/components/responses/Problem' }
components:
  securitySchemes:
    bearerAuth: { type: http, scheme: bearer, bearerFormat: JWT }
  schemas:
    MessageCreate:
      type: object
      required: [to, messages]
      additionalProperties: false
      properties:
        to:       { type: array, items: { type: string }, minItems: 1, maxItems: 500 }
        messages: { type: array, maxItems: 5, items: { $ref: '#/components/schemas/LineMessage' } }
        scheduled_at: { type: string, format: date-time }
    LineMessage:
      type: object
      required: [type, text]
      properties:
        type: { type: string, enum: [text] }
        text: { type: string, maxLength: 5000 }
    Message:
      type: object
      properties:
        id: { type: string }
        status: { type: string, enum: [queued, sending, completed, failed, canceled] }
  responses:
    Problem:
      description: Error
      content:
        application/problem+json:
          schema: { type: object }

冪等性、部分更新與並發控制 深入 高頻

場景：client 呼叫 POST /v1/messages 推播給 10 萬人，等了 30 秒 timeout。它該重送嗎？若伺服器其實已經收到並開始發送，重送會讓 10 萬人收到兩次。解法是讓 client 產生唯一的 Idempotency-Key（通常是 UUIDv4），伺服器以它為鍵記住「這個請求的結果」，重複請求直接回放同一個結果。這是 Stripe 推廣、IETF 正在標準化（draft-ietf-httpapi-idempotency-key-header）的做法。

設計要點

1. 鍵的範圍：去重鍵是 (tenant_id, idempotency_key)，避免不同客戶的 key 碰撞。
2. 請求指紋：同時儲存 request body 的 hash。同 key 但不同 body → 回 422（client 誤用 key），不可默默回放。
3. 處理中狀態：第一個請求還沒完成時第二個進來 → 回 409 Conflict（或等待），不能兩個都執行。靠 DB 唯一索引或 Redis SET NX 取得「鎖」。
4. 原子性：理想上「寫業務資料」與「寫冪等紀錄」在同一個 DB transaction；跨系統時（例如送到 Kafka）至少要保證業務動作本身可重入。
5. 失敗結果：4xx（驗證錯誤）可以快取回放；5xx 通常不快取，讓 client 用同一 key 重試能真的重新執行。
6. TTL：保留 24 小時到數天（Stripe 為 24 小時），之後清除；太短則延遲重試會重複執行。
7. 只對非冪等方法要求：POST（以及非冪等 PATCH）。GET/PUT/DELETE 本身已冪等。

CREATE TABLE idempotency_keys (
  tenant_id       BIGINT       NOT NULL,
  idem_key        VARCHAR(64)  NOT NULL,
  request_hash    CHAR(64)     NOT NULL,
  status          ENUM('processing','done') NOT NULL,
  response_code   SMALLINT     NULL,
  response_body   JSON         NULL,
  created_at      TIMESTAMP    NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (tenant_id, idem_key),
  KEY idx_created (created_at)          -- 給定期清理 job 用
);

import hashlib, json
from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse
import redis.asyncio as redis

app = FastAPI()
r = redis.Redis()
TTL = 24 * 3600

async def with_idempotency(tenant: str, key: str, body: bytes, handler):
    """Redis 版：SET NX 搶占，完成後覆寫成結果。"""
    rkey = f"idem:{tenant}:{key}"
    req_hash = hashlib.sha256(body).hexdigest()
    placeholder = json.dumps({"state": "processing", "hash": req_hash})
    acquired = await r.set(rkey, placeholder, nx=True, ex=TTL)
    if not acquired:
        saved = json.loads(await r.get(rkey) or "{}")
        if saved.get("hash") != req_hash:
            raise HTTPException(422, "Idempotency-Key reused with different payload")
        if saved.get("state") == "processing":
            raise HTTPException(409, "Request with this key is still in progress")
        return JSONResponse(saved["body"], status_code=saved["code"],
                            headers={"Idempotent-Replayed": "true"})
    try:
        code, payload = await handler()
    except Exception:
        await r.delete(rkey)          # 5xx 不快取：讓 client 能重試
        raise
    await r.set(rkey, json.dumps({"state": "done", "hash": req_hash,
                                  "code": code, "body": payload}), ex=TTL)
    return JSONResponse(payload, status_code=code)

@app.post("/v1/messages")
async def send_message(request: Request,
                       idempotency_key: str = Header(..., alias="Idempotency-Key"),
                       x_tenant_id: str = Header(...)):
    body = await request.body()
    async def handler():
        # ... 驗證、寫 DB、丟進佇列 ...
        return 202, {"id": "msg_123", "status": "queued"}
    return await with_idempotency(x_tenant_id, idempotency_key, body, handler)

部分更新 PATCH 與並發控制（ETag / If-Match） 進階

JSON Merge Patch vs JSON Patch

// Merge Patch：把預算改成 500、刪除 end_date、targeting.countries 整個換成 ["TW","JP"]
{ "budget": 500, "end_date": null, "targeting": { "countries": ["TW", "JP"] } }

// JSON Patch：只在 countries 尾端加一個，並確認目前 budget 仍為 300 才改
[
  { "op": "test",    "path": "/budget", "value": 300 },
  { "op": "replace", "path": "/budget", "value": 500 },
  { "op": "add",     "path": "/targeting/countries/-", "value": "KR" }
]

def merge_patch(target, patch):
    """RFC 7396 參考實作。"""
    if not isinstance(patch, dict):
        return patch                       # 非物件：整個取代（包含陣列）
    if not isinstance(target, dict):
        target = {}
    result = dict(target)
    for k, v in patch.items():
        if v is None:
            result.pop(k, None)            # null = 刪除
        else:
            result[k] = merge_patch(result.get(k), v)
    return result

assert merge_patch({"a": 1, "b": {"c": 2, "d": 3}}, {"b": {"c": None, "e": 4}}) == {"a": 1, "b": {"d": 3, "e": 4}}

ETag + If-Match 樂觀鎖

Lost update 問題：A、B 同時 GET 廣告活動（budget=300），A 改成 500、B 改名稱並以 PUT 送出完整物件（budget 仍是 300），A 的修改就被蓋掉了。解法：

-- 伺服器端以版本欄位實作：一條 SQL 完成「比較並設定」
UPDATE campaigns
SET budget = 500, version = version + 1
WHERE id = 1 AND version = 7;
-- affected rows = 0 → 版本已變 → 回 412

• ETag 可以是版本號或內容 hash；strong ETag 表示位元組完全相同，weak ETag（W/"v7"）表示語意相同，If-Match 要求 strong 比對。
• 同一個機制用在讀取：If-None-Match: "v7" 相符時回 304 Not Modified，省頻寬。
• 可要求所有更新必須帶 If-Match，沒帶回 428 Precondition Required。
• 樂觀鎖適合衝突少的場景；衝突頻繁時（例如熱門計數器）改用原子操作（UPDATE ... SET x = x + 1、Redis INCR）或悲觀鎖。

長時間任務、Webhook 與批次 API 進階 高頻

202 Accepted + Polling

任何超過一兩秒的工作（產生報表、大量推播、匯入檔案）都不該讓 HTTP 連線掛著等：會占用 worker、撞 LB/gateway timeout（AWS ALB idle timeout 預設 60 秒）、client 斷線就白做。標準模式：

POST /v1/report-exports
Content-Type: application/json

{"campaign_ids": ["cmp_1","cmp_2"], "from": "2026-10-01", "to": "2026-10-31", "format": "csv"}

HTTP/1.1 202 Accepted
Location: /v1/report-exports/exp_77
Retry-After: 10

{"id": "exp_77", "status": "pending", "created_at": "2026-11-03T10:00:00Z"}

GET /v1/report-exports/exp_77

HTTP/1.1 200 OK
{"id": "exp_77", "status": "succeeded", "progress": 1.0,
 "result": {"download_url": "https://bucket.s3.amazonaws.com/...", "expires_at": "2026-11-03T11:00:00Z"}}

Polling 的缺點是浪費請求；可用 Retry-After 指導頻率、或提供 long polling，最好的是讓 client 選擇訂閱 webhook。

Webhook 設計

Webhook 是「伺服器主動 HTTP POST 到 client 註冊的 URL」。設計時要假設接收方會慢、會掛、會回錯，而且有人會偽造請求。

1. 簽章驗證（HMAC）：用每個訂閱者專屬 secret 對 timestamp + "." + raw_body 做 HMAC-SHA256，放在 header（例如 Webhook-Signature: t=1730628000,v1=...）。接收端用原始 bytes（不是重新序列化後的 JSON）重算並用常數時間比對；同時檢查 timestamp 在 5 分鐘內，防重放攻擊。LINE Messaging API 的 webhook 也是用 channel secret 對 body 做 HMAC-SHA256，放在 x-line-signature（base64）。
2. At-least-once + 冪等：每個事件帶唯一 event_id；接收端用它去重。發送端不保證只送一次。
3. 重試與退避：非 2xx 或 timeout（例如 10 秒）就以指數退避重試數十次、跨越一到三天；持續失敗則自動停用端點並通知。
4. 快速 ack：要求接收端先回 2xx，再非同步處理，避免在 webhook handler 中做重工作。
5. 順序不保證：事件可能亂序到達。payload 帶 occurred_at 或版本號，接收端只接受較新的狀態；或只送「事件通知（thin payload）」讓接收端回頭 GET 最新狀態。
6. 安全：只允許 HTTPS；註冊 URL 時防 SSRF（禁止內網 IP、metadata endpoint 169.254.169.254）；提供 secret rotation（同時接受新舊兩把）。
7. 可觀測與重播：提供投遞紀錄 UI/API 與手動 replay。

import hmac, hashlib, time

def sign(secret: bytes, body: bytes, ts: int | None = None) -> str:
    ts = ts or int(time.time())
    mac = hmac.new(secret, f"{ts}.".encode() + body, hashlib.sha256).hexdigest()
    return f"t={ts},v1={mac}"

def verify(secret: bytes, body: bytes, header: str, tolerance: int = 300) -> bool:
    parts = dict(p.split("=", 1) for p in header.split(","))
    ts = int(parts["t"])
    if abs(time.time() - ts) > tolerance:
        return False                      # 防重放
    expected = hmac.new(secret, f"{ts}.".encode() + body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, parts["v1"])   # 常數時間比較，防 timing attack

# 驗證 LINE webhook（官方規格：HMAC-SHA256(channel_secret, body) 再 base64）
import base64
from fastapi import FastAPI, Request, Header, HTTPException

app = FastAPI()
CHANNEL_SECRET = b"..."

@app.post("/callbacks/line")
async def line_callback(request: Request, x_line_signature: str = Header(...)):
    body = await request.body()           # 必須用原始 bytes
    digest = hmac.new(CHANNEL_SECRET, body, hashlib.sha256).digest()
    if not hmac.compare_digest(base64.b64encode(digest).decode(), x_line_signature):
        raise HTTPException(401, "bad signature")
    # 先 enqueue 再立即回 200，處理交給 worker
    return {"ok": True}

批次 API

當 client 需要一次建立/更新上千筆資料時，逐筆呼叫會被往返延遲與 rate limit 拖垮。常見設計：

• POST /v1/audiences/aud_1/members:batchAdd，body 含最多 N 筆（例如 1000），超過回 413。
• 部分成功的語意要先定義：全有全無（transactional）或逐筆結果。逐筆時回 200 或 207 Multi-Status，body 對每筆給結果：

{
  "results": [
    { "index": 0, "status": 201, "id": "mem_1" },
    { "index": 1, "status": 422, "error": { "code": "invalid_email" } }
  ],
  "summary": { "succeeded": 1, "failed": 1 }
}

• 大量資料改用「上傳檔案到物件儲存 → 建立 import job → 202 → 輪詢/webhook」。
• Rate limit 以「筆數」而非「請求數」計算，避免批次被當成繞過限制的手段。

Rate Limiting 演算法與實作 深入 高頻

Rate limiting 的目的：保護服務不被打爆（包含自家 bug 造成的 retry storm）、公平分配資源給多租戶、控制成本、作為商業方案的配額、以及遵守上游（例如 LINE API）的限制。

Python 實作

import time
from collections import deque

class FixedWindow:
    def __init__(self, limit: int, window: float):
        self.limit, self.window = limit, window
        self.cur_window, self.count = 0, 0

    def allow(self, now: float | None = None) -> bool:
        now = time.monotonic() if now is None else now
        w = int(now // self.window)
        if w != self.cur_window:
            self.cur_window, self.count = w, 0
        if self.count < self.limit:
            self.count += 1
            return True
        return False

class SlidingLog:
    def __init__(self, limit: int, window: float):
        self.limit, self.window = limit, window
        self.log: deque[float] = deque()

    def allow(self, now: float | None = None) -> bool:
        now = time.monotonic() if now is None else now
        while self.log and self.log[0] <= now - self.window:
            self.log.popleft()
        if len(self.log) < self.limit:
            self.log.append(now)
            return True
        return False

class SlidingWindowCounter:
    def __init__(self, limit: int, window: float):
        self.limit, self.window = limit, window
        self.cur_start, self.cur, self.prev = 0.0, 0, 0

    def allow(self, now: float | None = None) -> bool:
        now = time.monotonic() if now is None else now
        start = (now // self.window) * self.window
        if start != self.cur_start:
            # 前進一格：若跳過不只一格，上一窗口視為 0
            self.prev = self.cur if start - self.cur_start == self.window else 0
            self.cur, self.cur_start = 0, start
        overlap = 1 - (now - start) / self.window          # 上一窗口仍在範圍內的比例
        estimated = self.prev * overlap + self.cur
        if estimated < self.limit:
            self.cur += 1
            return True
        return False

class TokenBucket:
    def __init__(self, rate: float, capacity: float):
        self.rate, self.capacity = rate, capacity
        self.tokens, self.last = capacity, time.monotonic()

    def allow(self, cost: float = 1.0) -> tuple[bool, float]:
        now = time.monotonic()
        self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate)
        self.last = now
        if self.tokens >= cost:
            self.tokens -= cost
            return True, 0.0
        retry_after = (cost - self.tokens) / self.rate     # 還要等多久才有足夠 token
        return False, retry_after

class LeakyBucket:
    """以「下一個可處理時間」實作（GCRA 的簡化），不需要真正的佇列。"""
    def __init__(self, rate: float, capacity: int):
        self.interval = 1.0 / rate           # 每個請求的間隔
        self.max_delay = capacity * self.interval
        self.next_free = 0.0

    def allow(self) -> tuple[bool, float]:
        now = time.monotonic()
        start = max(now, self.next_free)
        delay = start - now
        if delay > self.max_delay:
            return False, delay - self.max_delay
        self.next_free = start + self.interval
        return True, delay                   # 呼叫端可 sleep(delay) 達成平滑輸出

分散式 rate limit：Redis + Lua

多台 API server 時，計數必須放在共享儲存。用 Lua script 讓「讀取—計算—寫回」在 Redis 內原子執行，避免 race condition：

import time, redis

TOKEN_BUCKET_LUA = """
local key      = KEYS[1]
local rate     = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local now      = tonumber(ARGV[3])
local cost     = tonumber(ARGV[4])
local data   = redis.call('HMGET', key, 'tokens', 'ts')
local tokens = tonumber(data[1]) or capacity
local ts     = tonumber(data[2]) or now
tokens = math.min(capacity, tokens + (now - ts) * rate)
local allowed = 0
local retry   = 0
if tokens >= cost then
  tokens = tokens - cost
  allowed = 1
else
  retry = (cost - tokens) / rate
end
redis.call('HSET', key, 'tokens', tokens, 'ts', now)
redis.call('EXPIRE', key, math.ceil(capacity / rate) + 1)
return {allowed, tostring(retry), tostring(tokens)}
"""

r = redis.Redis()
bucket = r.register_script(TOKEN_BUCKET_LUA)

def check(tenant: str, rate=100, capacity=200, cost=1):
    allowed, retry, remaining = bucket(keys=[f"rl:{tenant}"],
                                       args=[rate, capacity, time.time(), cost])
    return bool(allowed), float(retry), float(remaining)

429 與 Retry-After

HTTP/1.1 429 Too Many Requests
Content-Type: application/problem+json
Retry-After: 3
RateLimit-Policy: "default";q=100;w=60
RateLimit: "default";r=0;t=3

{"type":"https://api.example.com/problems/rate-limited","title":"Too Many Requests","status":429,"detail":"Limit 100 requests per 60s exceeded"}

• Retry-After 可以是秒數或 HTTP 日期（RFC 9110）。
• 許多 API 使用 X-RateLimit-Limit / X-RateLimit-Remaining / X-RateLimit-Reset（GitHub 風格）；IETF 正在標準化 RateLimit 與 RateLimit-Policy header，草案語法仍在演進，實作前請查最新版本。
• Client 端應遵守 Retry-After，沒有時用指數退避 + jitter，避免所有 client 同時重試造成驚群（thundering herd）。
• 限流維度：per API key、per user、per IP、per endpoint（貴的 endpoint 限更緊）、全域；多層同時存在時回報最先觸發的那一層。

驗證、授權與 API 安全 深入 高頻

先分清兩個詞：Authentication（認證，AuthN）回答「你是誰」；Authorization（授權，AuthZ）回答「你能做什麼」。401 對應認證失敗，403 對應授權失敗。

Session vs JWT

JWT 結構：base64url(header).base64url(payload).signature。Payload 只是編碼不是加密，任何人都能解開閱讀，所以不能放敏感資料。

JWT 撤銷問題的解法

1. 短效 access token + refresh token：access token 5–15 分鐘過期，refresh token 長效但存在伺服器可撤銷。撤銷後最多再被用到 access token 剩餘壽命。
2. Denylist：登出或被盜時把 token 的 jti 放進 Redis（TTL = token 剩餘壽命）。每個請求查一次 Redis——等於部分回到有狀態，但只需存「被撤銷的」而非全部。
3. Token version：使用者表存 token_version，JWT 帶此值，改密碼時版本 +1 讓舊 token 全部失效（仍需查表，但可快取）。
4. Refresh token rotation：每次刷新發新的 refresh token 並作廢舊的；若舊的被重複使用，視為被盜，撤銷整個 token family。

import jwt, time   # PyJWT

PRIVATE_KEY = open("private.pem").read()
PUBLIC_KEY = open("public.pem").read()

def issue(user_id: str, tenant: str, scopes: list[str]) -> str:
    now = int(time.time())
    claims = {"sub": user_id, "tenant": tenant, "scope": " ".join(scopes),
              "iss": "https://auth.example.com", "aud": "push-api",
              "iat": now, "exp": now + 900, "jti": f"{user_id}-{now}"}
    return jwt.encode(claims, PRIVATE_KEY, algorithm="RS256",
                      headers={"kid": "2026-11"})   # kid 支援金鑰輪替

def verify(token: str) -> dict:
    return jwt.decode(token, PUBLIC_KEY,
                      algorithms=["RS256"],          # 明確白名單，拒絕 none/HS256
                      audience="push-api",
                      issuer="https://auth.example.com",
                      options={"require": ["exp", "iat", "sub"]})

OAuth 2.0 Flows

OAuth 2.0 是授權委派框架：讓第三方應用在不知道使用者密碼的情況下，取得有限範圍（scope）的存取權。它本身不是認證協定；在其上加一層 ID token 的是 OpenID Connect（OIDC）。四個角色：Resource Owner（使用者）、Client（第三方 App）、Authorization Server（發 token）、Resource Server（API）。

Authorization Code + PKCE

有使用者參與時的標準流程，適用 Web、SPA、行動 App。PKCE（Proof Key for Code Exchange）防止授權碼在重新導向過程被攔截後被拿去換 token：攔截者沒有原始的 code_verifier。OAuth 2.1 草案要求所有 client 都使用 PKCE。state 參數用來防 CSRF。

Client Credentials

沒有使用者，服務對服務（machine-to-machine）。Client 用 client_id + client_secret（或 mTLS、private key JWT）直接向 /token 換 access token。例如：報表排程服務呼叫內部資料 API。

Device Code

輸入不便的裝置（智慧電視、CLI）：裝置顯示一組代碼，使用者在手機上完成授權。

Refresh Token

用來換新的 access token，不需使用者再次登入。

已不建議：Implicit、Password（ROPC）

Implicit 把 token 放在 URL fragment 容易外洩；ROPC 讓 client 直接拿使用者密碼，違背 OAuth 的初衷。OAuth 2.1 已移除兩者。

import base64, hashlib, secrets

code_verifier = secrets.token_urlsafe(64)[:128]       # 43–128 字元
code_challenge = base64.urlsafe_b64encode(
    hashlib.sha256(code_verifier.encode()).digest()).decode().rstrip("=")
# authorize URL 帶 code_challenge & code_challenge_method=S256
# 換 token 時帶 code_verifier

# Client Credentials：服務對服務取得 token
curl -s -X POST https://auth.example.com/oauth/token \
  -u "$CLIENT_ID:$CLIENT_SECRET" \
  -d grant_type=client_credentials \
  -d scope="reports:read"

API Key 與 Scopes

• API Key：簡單、適合伺服器對伺服器的 B2B 整合（例如廣告主後台呼叫報表 API）。設計重點：只在建立時顯示一次；DB 只存 hash（SHA-256 足夠，因為 key 是高熵隨機值，不需 bcrypt）；加可辨識前綴（sk_live_、sk_test_）方便 secret scanning；支援多把 key 與輪替；可設定 scope、IP 白名單、到期日；記錄 last_used_at。
• Key 放在 Authorization: Bearer 或自訂 header，不要放 query string（會進 access log、瀏覽器歷史、Referer）。
• Scopes：粗粒度的權限宣告（messages:send、reports:read），由 token 攜帶，代表「這把 token 被允許做什麼」。Scope 與使用者本身的角色權限要取交集。

import hashlib, secrets

def create_api_key(env: str = "live") -> tuple[str, str]:
    raw = f"sk_{env}_{secrets.token_urlsafe(32)}"
    digest = hashlib.sha256(raw.encode()).hexdigest()
    return raw, digest        # raw 回給使用者一次；DB 只存 digest 與前 12 字元供顯示

def lookup(raw: str, db) -> dict | None:
    return db.get_key_by_hash(hashlib.sha256(raw.encode()).hexdigest())

RBAC vs ABAC

實務常見混合：RBAC 決定功能層級權限，再用屬性（尤其是 tenant_id、擁有者）做物件層級過濾。多租戶 SaaS 最重要的一條就是：每一個查詢都必須帶租戶條件，最好在資料存取層強制（例如 repository 自動加 WHERE tenant_id = :current_tenant，或 PostgreSQL Row-Level Security）。

API 安全：OWASP API Top 10、CORS、輸入驗證 進階

OWASP API Security Top 10（2023 版）重點

CORS

瀏覽器的同源政策（same-origin policy）預設阻擋 JS 讀取跨來源回應。CORS 是伺服器用 header 告訴瀏覽器「哪些來源可以讀」。重點：

• CORS 是瀏覽器強制的，對 curl、伺服器對伺服器呼叫無效，所以它不是授權機制。
• 非「簡單請求」（例如 Content-Type: application/json、自訂 header、PUT/DELETE）會先送 OPTIONS preflight。
• Access-Control-Allow-Origin: * 不能與 Access-Control-Allow-Credentials: true 並用；不要把請求的 Origin 原封不動反射回去（等於允許所有網站帶 cookie 讀你的 API）。
• 用 Access-Control-Max-Age 快取 preflight 結果，減少一次往返。

OPTIONS /v1/campaigns HTTP/1.1
Origin: https://dashboard.example.com
Access-Control-Request-Method: PATCH
Access-Control-Request-Headers: content-type, authorization

HTTP/1.1 204 No Content
Access-Control-Allow-Origin: https://dashboard.example.com
Access-Control-Allow-Methods: GET, POST, PATCH, DELETE
Access-Control-Allow-Headers: content-type, authorization, idempotency-key
Access-Control-Allow-Credentials: true
Access-Control-Max-Age: 600
Vary: Origin

輸入驗證原則

• 白名單優於黑名單：定義允許的型別、長度、格式、範圍、enum。
• 在邊界（API 層）用 schema 驗證（pydantic、JSON Schema），拒絕未知欄位（extra="forbid"）防 mass assignment。
• SQL 一律參數化查詢；動態排序欄位用白名單映射。
• 限制 body 大小、陣列長度、字串長度、巢狀深度（JSON bomb）。
• 輸出時依情境編碼（HTML escape 防 XSS）；驗證輸入不能取代輸出編碼。
• 所有敏感操作記錄 audit log（誰、何時、對什麼、做了什麼）。

GraphQL：Schema、Resolver、N+1 與 DataLoader 進階 高頻

GraphQL 是 Facebook 於 2012 年內部開發、2015 年開源的 API 查詢語言。核心概念：單一 endpoint（通常 POST /graphql），強型別 schema，由 client 決定要哪些欄位。它解決 REST 的兩個痛點：over-fetching（回了用不到的欄位）與 under-fetching（一個畫面要打 N 支 API）。

Schema 與三種操作

type Advertiser {
  id: ID!
  name: String!
  campaigns(first: Int = 20, after: String, status: CampaignStatus): CampaignConnection!
}

type Campaign {
  id: ID!
  name: String!
  status: CampaignStatus!
  budgetMicros: Int!          # 注意：GraphQL Int 是 32-bit，大數字要用自訂 scalar 或 String
  stats(from: Date!, to: Date!): CampaignStats!
}

enum CampaignStatus { ACTIVE PAUSED ENDED }

type CampaignConnection {     # Relay cursor connection 規格
  edges: [CampaignEdge!]!
  pageInfo: PageInfo!
}
type CampaignEdge { node: Campaign!  cursor: String! }
type PageInfo { hasNextPage: Boolean!  endCursor: String }

type Query {
  advertiser(id: ID!): Advertiser
}

type Mutation {
  pauseCampaign(id: ID!): PauseCampaignPayload!
}
type PauseCampaignPayload { campaign: Campaign  userErrors: [UserError!]! }
type UserError { field: [String!]  message: String! }

type Subscription {
  campaignStatusChanged(advertiserId: ID!): Campaign!
}

query AdvertiserDashboard($id: ID!) {
  advertiser(id: $id) {
    name
    campaigns(first: 10, status: ACTIVE) {
      edges { node { id name stats(from: "2026-10-01", to: "2026-10-31") { impressions clicks spendMicros } } }
      pageInfo { hasNextPage endCursor }
    }
  }
}

Query

唯讀查詢，同一層欄位可平行解析。

Mutation

寫入操作；頂層多個 mutation 欄位依序執行。慣例回傳 payload 型別，包含更新後物件與業務錯誤（userErrors）。

Subscription

伺服器推送事件，通常走 WebSocket（graphql-ws 協定）或 SSE。需要有狀態的長連線，水平擴展時要搭配 pub/sub（Redis、Kafka）。

Resolver

每個欄位對應一個函式 resolve(parent, args, context, info)，執行引擎從根開始逐層呼叫。context 放 request 範圍的東西（目前使用者、DB session、DataLoader 實例）。

錯誤處理

GraphQL 通常永遠回 HTTP 200，錯誤放在回應的 errors 陣列，且可能部分成功（data 有部分欄位、errors 指出哪個 path 失敗）。這讓以 status code 為基礎的監控失效，必須另外解析 body 或在伺服器端以 operation 為單位記錄錯誤率。

N+1 問題與 DataLoader

上面的查詢：取 1 個 advertiser → 取 10 個 campaigns → 對每個 campaign 呼叫 stats resolver。若每個 resolver 各自查 DB，就是 1 + 1 + 10 次查詢；巢狀更深時呈乘法成長。

DataLoader 的做法：在同一個 event loop tick 內收集所有 load(key) 呼叫，合併成一次 batch_load(keys)（例如 WHERE campaign_id IN (...)），並在 request 範圍內做 memoization（同一 key 只查一次）。

import asyncio
from collections import defaultdict

class DataLoader:
    """極簡版 DataLoader：同一輪 event loop 內的 load() 合併成一次批次查詢。"""
    def __init__(self, batch_fn):
        self.batch_fn = batch_fn          # async (list[key]) -> list[value]，順序需與 keys 對應
        self.cache: dict = {}
        self.queue: list = []

    def load(self, key):
        if key in self.cache:
            return self.cache[key]
        fut = asyncio.get_running_loop().create_future()
        self.cache[key] = fut
        self.queue.append((key, fut))
        if len(self.queue) == 1:
            asyncio.get_running_loop().call_soon(lambda: asyncio.ensure_future(self._dispatch()))
        return fut

    async def _dispatch(self):
        batch, self.queue = self.queue, []
        keys = [k for k, _ in batch]
        try:
            values = await self.batch_fn(keys)
            for (_, fut), v in zip(batch, values):
                fut.set_result(v)
        except Exception as e:
            for _, fut in batch:
                fut.set_exception(e)

async def batch_stats(campaign_ids):
    print("ONE query for", campaign_ids)
    # SELECT campaign_id, SUM(impressions) ... WHERE campaign_id IN (...) GROUP BY campaign_id
    rows = {cid: {"impressions": 100 * i} for i, cid in enumerate(campaign_ids)}
    return [rows.get(cid) for cid in campaign_ids]   # 必須依 keys 順序回傳，缺的補 None

async def main():
    loader = DataLoader(batch_stats)                  # 每個 request 建一個新的，避免跨使用者快取
    results = await asyncio.gather(*(loader.load(f"cmp_{i}") for i in range(10)))
    print(results[:2])

asyncio.run(main())

實務上 Python 用 Strawberry（內建 strawberry.dataloader.DataLoader）、Graphene + aiodataloader；Node 用官方 dataloader 套件。

保護 GraphQL 伺服器

• Query depth limit：限制巢狀深度（例如 10），防 friends { friends { friends ... } } } 這種指數爆炸。
• Query complexity / cost analysis：每個欄位給成本，列表欄位乘上 first 參數，總成本超過上限就在執行前拒絕。GitHub GraphQL API 以「點數」計算 rate limit 就是這個概念。
• 強制列表分頁上限：first 最大 100。
• Timeout 與 關閉 production introspection（視需求）。
• Persisted queries：client 在建置時把查詢註冊到伺服器、執行時只送 hash（{"id":"sha256:abc", "variables":{...}}）。好處：(1) 安全——伺服器只執行白名單查詢，等於封死任意查詢攻擊；(2) 請求更小；(3) 可以改用 GET 讓 CDN 快取。Apollo 的 APQ（Automatic Persisted Queries）是「先送 hash，伺服器沒看過才補送全文」的變體，主要省頻寬而非安全。

為什麼 GraphQL 快取困難？

• HTTP 快取以 URL 為鍵，而 GraphQL 全部是 POST /graphql，CDN/瀏覽器快取直接失效。
• 每個 client 查詢的欄位組合不同，回應難以共用。
• 解法：client 端正規化快取（Apollo Client 以 __typename + id 為鍵）、persisted queries + GET 讓 CDN 快取、伺服器端以 resolver/DataLoader 層快取、@cacheControl 指令計算整份回應的最短 max-age。

何時選 GraphQL？

gRPC 與 REST / GraphQL / gRPC 比較 進階

gRPC 是 Google 開源的 RPC 框架：用 Protocol Buffers 定義服務與訊息、用 HTTP/2 傳輸、自動產生多語言的 client/server stub。

syntax = "proto3";
package bidding.v1;

import "google/protobuf/timestamp.proto";

service Bidder {
  // Unary：一問一答
  rpc Bid(BidRequest) returns (BidResponse);
  // Server streaming：一個請求，伺服器持續回傳
  rpc WatchBudget(WatchBudgetRequest) returns (stream BudgetUpdate);
  // Client streaming：client 持續上傳，伺服器最後回一次
  rpc UploadEvents(stream Event) returns (UploadSummary);
  // Bidirectional streaming：雙向獨立的串流
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message BidRequest {
  string request_id = 1;
  string placement_id = 2;
  repeated string user_segments = 3;
  int64 floor_price_micros = 4;
  google.protobuf.Timestamp ts = 5;
  reserved 6;                 // 曾經使用過、已刪除的欄位號碼，永遠保留
  reserved "device_ip";
}

message BidResponse {
  string request_id = 1;
  bool no_bid = 2;
  int64 price_micros = 3;
  string creative_id = 4;
}

為什麼 gRPC 快？

• Protobuf 二進位編碼：欄位以「欄位號碼 + wire type」的 tag 標示，數字用 varint 編碼，不傳欄位名稱，體積通常明顯小於 JSON，序列化/反序列化也更省 CPU。
• HTTP/2：單一 TCP 連線上多路複用（multiplexing）多個 stream，避免 HTTP/1.1 的 head-of-line blocking（應用層）與重複建連；HPACK 壓縮 header；原生支援雙向串流。
• 強型別契約與 codegen：少了手寫序列化與解析錯誤。

Deadline 與錯誤碼

gRPC 鼓勵每個呼叫都設 deadline（絕對截止時間），而且 deadline 會沿呼叫鏈傳遞：A 給 B 100ms，B 已經用了 30ms，呼叫 C 時剩 70ms。超時回 DEADLINE_EXCEEDED，下游也能感知取消而停止工作，避免「上游早已放棄、下游還在做白工」。gRPC 預設沒有 deadline（無限等待），這是常見事故來源。

import grpc
from bidding.v1 import bidder_pb2, bidder_pb2_grpc

channel = grpc.insecure_channel("bidder:50051")   # 長連線重用，不要每次建立
stub = bidder_pb2_grpc.BidderStub(channel)
try:
    resp = stub.Bid(bidder_pb2.BidRequest(request_id="r1", placement_id="p9"),
                    timeout=0.08)                   # 80ms deadline
except grpc.RpcError as e:
    if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
        resp = bidder_pb2.BidResponse(request_id="r1", no_bid=True)   # 逾時就不出價
    else:
        raise

常見狀態碼：OK、INVALID_ARGUMENT、NOT_FOUND、ALREADY_EXISTS、PERMISSION_DENIED、UNAUTHENTICATED、RESOURCE_EXHAUSTED（限流）、FAILED_PRECONDITION、UNAVAILABLE（可重試）、DEADLINE_EXCEEDED、INTERNAL。

Protobuf 向後相容規則

1. 永遠不要改已使用欄位的號碼——wire format 只認號碼不認名字。
2. 刪除欄位時用 reserved 保留號碼與名稱，防止日後被重用造成舊資料被誤解。
3. 新增欄位是安全的：舊程式碼會忽略不認識的欄位（unknown fields）；新程式碼讀到舊資料時取得預設值（0、空字串、false）。
4. 不要改型別（少數相容組合例外，如 int32/int64/uint32 之間，但有截斷風險）。
5. proto3 的 scalar 欄位無法分辨「沒設定」與「預設值」，需要時用 optional 關鍵字或 wrapper types。
6. enum 第一個值必須是 0，慣例命名 XXX_UNSPECIFIED，作為未知值的落點。
7. 欄位號碼 1–15 編碼只需 1 byte tag，留給最常出現的欄位。

何時選 gRPC？

• 內部微服務之間、高 QPS、低延遲、多語言（例如 Python 模型服務 ↔ Go bidder）。
• 需要串流（即時事件、大量上傳）。
• 不適合直接給瀏覽器（需要 gRPC-Web + proxy，如 Envoy）、不適合要用 curl 除錯或公開給第三方的 API。L7 負載平衡要支援 HTTP/2，否則長連線會讓流量全黏在同一個後端（Kubernetes 預設 Service 是 L4，要用 client-side LB 或 service mesh）。

REST vs GraphQL vs gRPC 比較 基礎 高頻

FastAPI 實戰：Pydantic、依賴注入、Async 與錯誤處理 進階

FastAPI 建立在 Starlette（ASGI）與 Pydantic 之上，型別提示同時驅動驗證、序列化、文件與依賴注入。以下是一個可直接執行的完整範例（uv pip install fastapi uvicorn pydantic，uvicorn app:app --reload）。

# app.py  （Python 3.10+）
import uuid
from datetime import datetime, timezone
from enum import Enum
from typing import Annotated

from fastapi import Depends, FastAPI, Header, Query, Request, Response, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ConfigDict, Field, field_validator

app = FastAPI(title="Campaign API", version="1.0.0")

# ---------- Pydantic models：request 與 response 分開，避免 mass assignment ----------
class CampaignStatus(str, Enum):
    active = "active"
    paused = "paused"

class CampaignCreate(BaseModel):
    model_config = ConfigDict(extra="forbid")          # 拒絕未知欄位
    name: str = Field(min_length=1, max_length=100)
    budget_micros: int = Field(gt=0, le=10**13)
    countries: list[str] = Field(default_factory=list, max_length=50)

    @field_validator("countries")
    @classmethod
    def upper_iso(cls, v: list[str]) -> list[str]:
        if any(len(c) != 2 for c in v):
            raise ValueError("country must be ISO 3166-1 alpha-2")
        return [c.upper() for c in v]

class CampaignPatch(BaseModel):
    model_config = ConfigDict(extra="forbid")
    name: str | None = Field(default=None, min_length=1, max_length=100)
    budget_micros: int | None = Field(default=None, gt=0)
    status: CampaignStatus | None = None

class Campaign(BaseModel):
    id: str
    tenant_id: str
    name: str
    budget_micros: int
    countries: list[str]
    status: CampaignStatus
    version: int
    created_at: datetime

class Page(BaseModel):
    data: list[Campaign]
    next_cursor: str | None

# ---------- 自訂業務例外 + 統一 Problem Details ----------
class APIError(Exception):
    def __init__(self, status_code: int, type_: str, title: str, detail: str = ""):
        self.status_code, self.type, self.title, self.detail = status_code, type_, title, detail

def problem(request: Request, status_code: int, type_: str, title: str, detail="", **extra):
    body = {"type": f"https://api.example.com/problems/{type_}", "title": title,
            "status": status_code, "detail": detail, "instance": str(request.url.path),
            "request_id": getattr(request.state, "request_id", None), **extra}
    return JSONResponse(body, status_code=status_code, media_type="application/problem+json")

@app.exception_handler(APIError)
async def api_error_handler(request: Request, exc: APIError):
    return problem(request, exc.status_code, exc.type, exc.title, exc.detail)

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    errors = [{"field": ".".join(map(str, e["loc"][1:])), "message": e["msg"]} for e in exc.errors()]
    return problem(request, 422, "validation-error", "Request validation failed", errors=errors)

@app.exception_handler(Exception)
async def unhandled(request: Request, exc: Exception):
    # 詳細內容只寫 log，不回給 client
    return problem(request, 500, "internal", "Internal Server Error")

@app.middleware("http")
async def request_id_mw(request: Request, call_next):
    request.state.request_id = request.headers.get("X-Request-ID") or uuid.uuid4().hex
    resp = await call_next(request)
    resp.headers["X-Request-ID"] = request.state.request_id
    return resp

# ---------- 依賴注入：認證、租戶、儲存層 ----------
class Principal(BaseModel):
    user_id: str
    tenant_id: str
    scopes: set[str]

async def get_principal(authorization: Annotated[str | None, Header()] = None) -> Principal:
    if not authorization or not authorization.startswith("Bearer "):
        raise APIError(401, "unauthenticated", "Missing bearer token")
    # 真實情況：verify JWT；這裡示範用
    return Principal(user_id="u1", tenant_id="t1", scopes={"campaigns:read", "campaigns:write"})

def require_scope(scope: str):
    async def checker(p: Annotated[Principal, Depends(get_principal)]) -> Principal:
        if scope not in p.scopes:
            raise APIError(403, "forbidden", f"Missing scope {scope}")
        return p
    return checker

class CampaignRepo:
    """記憶體版 repository；正式環境換成 async DB 實作，測試時用 dependency_overrides 替換。"""
    def __init__(self):
        self.rows: dict[str, Campaign] = {}

    async def create(self, tenant: str, data: CampaignCreate) -> Campaign:
        c = Campaign(id=f"cmp_{uuid.uuid4().hex[:12]}", tenant_id=tenant, status=CampaignStatus.active,
                     version=1, created_at=datetime.now(timezone.utc), **data.model_dump())
        self.rows[c.id] = c
        return c

    async def get(self, tenant: str, cid: str) -> Campaign | None:
        c = self.rows.get(cid)
        return c if c and c.tenant_id == tenant else None    # 租戶隔離：防 BOLA

    async def list(self, tenant: str, limit: int, after: str | None) -> tuple[list[Campaign], str | None]:
        items = sorted((c for c in self.rows.values() if c.tenant_id == tenant), key=lambda c: c.id)
        if after:
            items = [c for c in items if c.id > after]
        page = items[: limit + 1]
        return page[:limit], (page[limit - 1].id if len(page) > limit else None)

_repo = CampaignRepo()
def get_repo() -> CampaignRepo:
    return _repo

Writer = Annotated[Principal, Depends(require_scope("campaigns:write"))]
Reader = Annotated[Principal, Depends(require_scope("campaigns:read"))]
Repo = Annotated[CampaignRepo, Depends(get_repo)]

# ---------- Endpoints ----------
@app.post("/v1/campaigns", status_code=status.HTTP_201_CREATED, response_model=Campaign)
async def create_campaign(body: CampaignCreate, p: Writer, repo: Repo, response: Response):
    c = await repo.create(p.tenant_id, body)
    response.headers["Location"] = f"/v1/campaigns/{c.id}"   # 201 搭配 Location
    return c

@app.get("/v1/campaigns", response_model=Page)
async def list_campaigns(p: Reader, repo: Repo,
                         limit: Annotated[int, Query(ge=1, le=100)] = 20,
                         cursor: str | None = None):
    items, nxt = await repo.list(p.tenant_id, limit, cursor)
    return Page(data=items, next_cursor=nxt)

@app.get("/v1/campaigns/{cid}", response_model=Campaign)
async def get_campaign(cid: str, p: Reader, repo: Repo):
    c = await repo.get(p.tenant_id, cid)
    if not c:
        raise APIError(404, "not-found", "Campaign not found")
    return JSONResponse(c.model_dump(mode="json"), headers={"ETag": f'"{c.version}"'})

@app.patch("/v1/campaigns/{cid}", response_model=Campaign)
async def patch_campaign(cid: str, body: CampaignPatch, p: Writer, repo: Repo,
                         if_match: Annotated[str | None, Header()] = None):
    c = await repo.get(p.tenant_id, cid)
    if not c:
        raise APIError(404, "not-found", "Campaign not found")
    if if_match is None:
        raise APIError(428, "precondition-required", "If-Match header is required")
    if if_match.strip('"') != str(c.version):
        raise APIError(412, "version-conflict", "Resource was modified", f"current version {c.version}")
    changes = body.model_dump(exclude_unset=True)     # 只套用 client 有送的欄位
    updated = c.model_copy(update={**changes, "version": c.version + 1})
    repo.rows[cid] = updated
    return JSONResponse(updated.model_dump(mode="json"), headers={"ETag": f'"{updated.version}"'})

重點解說

• Request/Response model 分離：CampaignCreate 不含 id、tenant_id、status，client 無法注入；response_model 過濾輸出欄位，防止洩漏內部欄位（OWASP API3）。
• 依賴注入：Depends 讓認證、授權、DB session、設定可組合、可快取（同一 request 內同一 dependency 只執行一次），且測試時可用 app.dependency_overrides[get_repo] = lambda: FakeRepo() 替換。
• async 的陷阱：async def endpoint 在 event loop 上執行，若裡面呼叫同步阻塞 I/O（requests.get、同步 DB driver、time.sleep）會卡住整個 worker 的所有請求。要嘛用 async 函式庫（httpx、asyncpg、redis.asyncio），要嘛把 endpoint 宣告成一般 def——FastAPI 會把它丟到 threadpool 執行。CPU 密集工作則應移到 process pool 或背景 worker。
• 錯誤處理：自訂例外 + 全域 handler 統一成 Problem Details；未捕捉例外只回通用訊息與 request id。
• 部署：uvicorn 或 gunicorn -k uvicorn.workers.UvicornWorker -w N，N 通常從 CPU 核心數開始調整；前面放 Nginx/LB 處理 TLS、緩衝慢速 client。

設計演練：發送 LINE 訊息的推播 API 深入 高頻

這是 Appier 實際出過的 take-home 題型。面試官想看的不是你會不會呼叫 LINE SDK，而是能不能把一個「看起來只是轉發 HTTP 請求」的功能，設計成可靠、可擴展、可觀測、多租戶的服務。下面用面試時的答題順序展開。

1. 釐清需求

2. 資源設計與 Endpoints

3. Request / Response

POST /v1/messages HTTP/1.1
Authorization: Bearer eyJhbGciOi...
Idempotency-Key: 6f1c2b8e-1d7a-4c39-9f0b-3a5c2e9d1b77
Content-Type: application/json

{
  "channel_id": "ch_01HF3",
  "audience": {
    "type": "user_ids",
    "user_ids": ["U4af4980629f0f0f0f0f0f0f0f0f0f0f0", "Ub1c2d3e4f5a6b7c8d9e0f1a2b3c4d5e6"]
  },
  "template_id": "tpl_black_friday",
  "variables": {
    "default": { "coupon": "BF2026" },
    "per_user": { "U4af4980629f0f0f0f0f0f0f0f0f0f0f0": { "first_name": "Amy" } }
  },
  "schedule_at": "2026-11-27T12:00:00+08:00",
  "options": { "notification_disabled": false, "expire_after_seconds": 86400 },
  "metadata": { "campaign_id": "cmp_77" }
}

HTTP/1.1 202 Accepted
Location: /v1/messages/msg_01HF9Z
Content-Type: application/json

{
  "id": "msg_01HF9Z",
  "status": "scheduled",
  "channel_id": "ch_01HF3",
  "audience_size": 2,
  "schedule_at": "2026-11-27T04:00:00Z",
  "created_at": "2026-11-03T10:00:00Z"
}

{
  "id": "msg_01HF9Z",
  "status": "sending",
  "stats": { "total": 1000000, "queued": 420000, "sent": 575000, "failed": 5000, "canceled": 0 },
  "failure_breakdown": { "user_blocked_or_invalid": 4800, "upstream_error": 200 },
  "started_at": "2026-11-27T04:00:01Z",
  "estimated_completion_at": "2026-11-27T04:12:00Z"
}

audience.type 也可以是 segment（例如 {"type":"segment","segment_id":"seg_vip"}，在 worker 端才展開成名單），避免 client 一次上傳百萬 ID；或是 broadcast。狀態機：scheduled → queued → sending → completed | partially_failed | failed | canceled。

4. 架構

1. API 層只做輕量工作：驗證、檢查冪等、寫一筆 message 與 outbox 事件（同一個 transaction），立即回 202。不在請求內呼叫 LINE。
2. Transactional Outbox：避免「DB 寫入成功但 publish 到佇列失敗」的雙寫不一致；relay 讀 outbox 送到佇列（Kafka / SQS / RabbitMQ 皆可，見第 11 章）。
3. Fan-out worker：把分群展開成收件者清單，切成 500 人一批（multicast 上限），每批產生確定性的 batch_id = hash(message_id, batch_no)，作為後續去重與 LINE 重試鍵的來源。
4. Sender worker：依 channel 取 token bucket 配額後呼叫 LINE；結果寫回 status store，失敗依錯誤類型決定重試或終止。
5. Scheduler：定期掃描 schedule_at <= now AND status = 'scheduled'（需索引），用 SELECT ... FOR UPDATE SKIP LOCKED 讓多個 scheduler 實例安全分工。

5. 重試、退避與錯誤分類

LINE Messaging API 提供 X-Line-Retry-Key header（UUID）：同一個 key 重送時 LINE 不會重複發送，而是回 409 表示已接受過。把它設成由 batch_id 決定的 UUID（例如 uuid5），即使 worker 在「LINE 已收到、我們還沒寫回結果」時 crash，重新處理也不會重複推播。這把 at-least-once 的佇列投遞變成了效果上的 exactly-once。

import asyncio, random, uuid, time
import httpx

LINE_MULTICAST = "https://api.line.me/v2/bot/message/multicast"
NAMESPACE = uuid.UUID("8a7c1f00-0000-4000-8000-000000000000")

class Retryable(Exception):
    def __init__(self, retry_after: float | None = None):
        self.retry_after = retry_after

class Permanent(Exception):
    pass

def backoff(attempt: int, base: float = 0.5, cap: float = 60.0) -> float:
    """Full jitter：在 [0, min(cap, base*2^attempt)] 之間均勻取值，避免同步重試造成驚群。"""
    return random.uniform(0, min(cap, base * (2 ** attempt)))

class ChannelLimiter:
    """單機版 token bucket；多台 worker 時改用前面的 Redis + Lua 版本。"""
    def __init__(self, rate: float, burst: float):
        self.rate, self.burst = rate, burst
        self.tokens, self.last = burst, time.monotonic()
        self.lock = asyncio.Lock()

    async def acquire(self):
        async with self.lock:
            while True:
                now = time.monotonic()
                self.tokens = min(self.burst, self.tokens + (now - self.last) * self.rate)
                self.last = now
                if self.tokens >= 1:
                    self.tokens -= 1
                    return
                await asyncio.sleep((1 - self.tokens) / self.rate)

    def slow_down(self, factor: float = 0.5):
        self.rate = max(1.0, self.rate * factor)     # 收到 429 時自適應降速（AIMD 的 MD）

async def send_batch(client: httpx.AsyncClient, token: str, batch_id: str,
                     user_ids: list[str], messages: list[dict]) -> str:
    retry_key = str(uuid.uuid5(NAMESPACE, batch_id))  # 同一批永遠同一把 key
    resp = await client.post(
        LINE_MULTICAST,
        headers={"Authorization": f"Bearer {token}", "X-Line-Retry-Key": retry_key},
        json={"to": user_ids, "messages": messages},
        timeout=httpx.Timeout(10.0, connect=3.0),
    )
    if resp.status_code == 200:
        return "sent"
    if resp.status_code == 409:
        return "sent"                                # 已被接受過：冪等重試成功
    if resp.status_code == 429:
        ra = resp.headers.get("Retry-After")
        raise Retryable(float(ra) if ra else None)
    if resp.status_code >= 500:
        raise Retryable()
    raise Permanent(f"{resp.status_code}: {resp.text[:200]}")

async def process(job: dict, client: httpx.AsyncClient, limiter: ChannelLimiter,
                  max_attempts: int = 8) -> str:
    for attempt in range(max_attempts):
        await limiter.acquire()
        try:
            return await send_batch(client, job["token"], job["batch_id"],
                                    job["user_ids"], job["messages"])
        except Retryable as e:
            if e.retry_after is not None:
                limiter.slow_down()
            delay = e.retry_after if e.retry_after is not None else backoff(attempt)
            await asyncio.sleep(delay)
            # 正式環境：不要在 worker 內長時間 sleep 占住並行度，
            # 而是把 job 帶著 attempt+1 丟進延遲佇列（SQS delay / Redis ZSET / Kafka retry topic）
        except (httpx.TimeoutException, httpx.TransportError):
            await asyncio.sleep(backoff(attempt))
        except Permanent:
            return "failed"
    return "dead_lettered"

6. Rate limit 的兩個方向

• 對下游（我們的 client）：API Gateway 依租戶方案限制每秒建立任務數與每日收件者總數，回 429 + Retry-After。這保護的是我們自己的系統。
• 對上游（LINE）：LINE 的限制是以 channel 為單位，所以 token bucket 的鍵是 channel_id，且必須在所有 sender worker 之間共享（Redis）。也要追蹤每月配額，接近上限時提前通知租戶，而不是發到一半失敗。
• 公平性（noisy neighbor）：某租戶一次丟 1,000 萬人，不能讓其他租戶的交易型訊息（例如 OTP、訂單通知）排在後面等一小時。做法：依租戶或 channel 分區/分佇列、加權公平排程（weighted round robin）、交易型與行銷型訊息分不同優先級佇列。

7. 多租戶與資料設計

CREATE TABLE messages (
  id              CHAR(26)     PRIMARY KEY,          -- ULID：時間有序，利於索引與 keyset 分頁
  tenant_id       BIGINT       NOT NULL,
  channel_id      CHAR(26)     NOT NULL,
  idem_key        VARCHAR(64)  NOT NULL,
  request_hash    CHAR(64)     NOT NULL,
  status          VARCHAR(20)  NOT NULL,
  audience        JSON         NOT NULL,
  schedule_at     DATETIME(3)  NULL,
  total           INT          NOT NULL DEFAULT 0,
  sent            INT          NOT NULL DEFAULT 0,
  failed          INT          NOT NULL DEFAULT 0,
  created_at      DATETIME(3)  NOT NULL,
  UNIQUE KEY uk_idem (tenant_id, idem_key),          -- 冪等去重
  KEY idx_tenant_created (tenant_id, created_at, id),
  KEY idx_schedule (status, schedule_at)
);

-- 收件者層級狀態量很大（每則活動可達千萬筆），可放在依 message_id 分區的表、
-- 或寫進 NoSQL / 欄式儲存；API 只提供分頁查詢與彙總統計。
CREATE TABLE message_batches (
  message_id   CHAR(26)    NOT NULL,
  batch_no     INT         NOT NULL,
  status       VARCHAR(20) NOT NULL,
  attempts     SMALLINT    NOT NULL DEFAULT 0,
  last_error   VARCHAR(255),
  updated_at   DATETIME(3) NOT NULL,
  PRIMARY KEY (message_id, batch_no)
);

• 隔離：所有查詢帶 tenant_id；channel access token 以 KMS 加密存放，worker 執行時才解密，log 中遮罩。
• 統計計數：百萬收件者的 sent/failed 計數若每批都 UPDATE messages SET sent = sent + 500 會形成熱點列；改為 Redis HINCRBY 累計、定期回寫 DB，或從事件流彙總。
• 可觀測：metrics 以 channel、result 為標籤（不要用 user id 當標籤，避免 cardinality 爆炸），追蹤佇列延遲、LINE 回應延遲、429 比例、DLQ 深度（見第 9 章）。

設計演練：合併兩個 JSON 檔，並把它 API 化 進階 高頻

題目常見形式：「有兩個 JSON 檔，各是 List[dict]，以 id 關聯，請合併成一份」，接著追問「欄位衝突怎麼辦？」「資料量到上億筆怎麼辦？」「如果要做成 API？」

先問清楚的問題

• 合併語意：inner join（兩邊都有才保留）、left join、還是 full outer join（聯集）？
• 同一欄位兩邊值不同時以誰為準？（固定優先、較新的 updated_at 為準、或保留兩者）
• 巢狀物件要深層合併還是整個覆蓋？陣列要聯集、串接還是取代？
• 同一個檔案內有重複 id 嗎？缺 id 的紀錄怎麼處理？
• 輸出順序有要求嗎？檔案多大？

記憶體內版本：O(n + m)

import json
from typing import Any, Callable, Iterable, Literal

Policy = Literal["left", "right", "newest"]

def deep_merge(a: Any, b: Any, prefer: Callable[[Any, Any], Any]) -> Any:
    if isinstance(a, dict) and isinstance(b, dict):
        out = dict(a)
        for k, v in b.items():
            out[k] = deep_merge(a[k], v, prefer) if k in a else v
        return out
    if isinstance(a, list) and isinstance(b, list):
        merged, seen = [], set()
        for x in a + b:                      # 陣列：保序去重聯集（元素需可序列化）
            key = json.dumps(x, sort_keys=True)
            if key not in seen:
                seen.add(key)
                merged.append(x)
        return merged
    return prefer(a, b)

def merge_records(left: Iterable[dict], right: Iterable[dict], key: str = "id",
                  how: Literal["inner", "left", "outer"] = "outer",
                  policy: Policy = "right") -> list[dict]:
    def index(rows: Iterable[dict]) -> dict:
        idx: dict = {}
        for r in rows:
            if key not in r:
                continue                      # 或收集到 errors 回報
            k = r[key]
            idx[k] = deep_merge(idx[k], r, lambda x, y: y) if k in idx else r   # 同檔重複 id 先自我合併
        return idx

    L, R = index(left), index(right)

    def prefer(row_l: dict, row_r: dict):
        if policy == "left":
            return lambda x, y: x
        if policy == "right":
            return lambda x, y: y
        newer_right = row_r.get("updated_at", "") >= row_l.get("updated_at", "")
        return (lambda x, y: y) if newer_right else (lambda x, y: x)

    if how == "inner":
        keys = [k for k in L if k in R]
    elif how == "left":
        keys = list(L)
    else:
        keys = list(L) + [k for k in R if k not in L]   # 保留左邊順序，再接右邊新增的

    out = []
    for k in keys:
        if k in L and k in R:
            out.append(deep_merge(L[k], R[k], prefer(L[k], R[k])))
        else:
            out.append(L[k] if k in L else R[k])
    return out

if __name__ == "__main__":
    a = [{"id": 1, "name": "Amy", "tags": ["vip"], "profile": {"age": 30}},
         {"id": 2, "name": "Bob"}]
    b = [{"id": 1, "email": "amy@example.com", "tags": ["tw"], "profile": {"city": "Taipei"}},
         {"id": 3, "name": "Cat"}]
    print(json.dumps(merge_records(a, b), ensure_ascii=False, indent=2))
    # id=1 → name, email, tags=["vip","tw"], profile={"age":30,"city":"Taipei"}；id=2、id=3 原樣保留

時間複雜度 O(n + m)（以 dict 建索引，假設合併單筆的成本為常數），空間 O(n + m)。面試時要主動說出「用 hash map 建索引，避免 O(n × m) 的雙層迴圈」。

資料量到上億筆怎麼辦？

import heapq, json
from typing import Iterator

def read_sorted_jsonl(path: str) -> Iterator[dict]:
    with open(path) as f:
        for line in f:
            yield json.loads(line)

def merge_join_sorted(path_a: str, path_b: str, key: str = "id") -> Iterator[dict]:
    """兩個檔案都已依 key 排序（JSON Lines），記憶體 O(1)，一次循序掃描。"""
    ia, ib = read_sorted_jsonl(path_a), read_sorted_jsonl(path_b)
    a, b = next(ia, None), next(ib, None)
    while a is not None or b is not None:
        if b is None or (a is not None and a[key] < b[key]):
            yield a; a = next(ia, None)
        elif a is None or b[key] < a[key]:
            yield b; b = next(ib, None)
        else:
            yield {**a, **b}                         # 依政策合併
            a, b = next(ia, None), next(ib, None)

# 外部排序的 k-way merge 可以直接用 heapq.merge(*sorted_chunk_iterators, key=lambda r: r["id"])

把它 API 化

依資料大小分兩種模式：

• 小資料、同步：POST /v1/merges，body {"left": [...], "right": [...], "key": "id", "how": "outer", "conflict_policy": "newest"}，限制 body 大小（例如數 MB）與筆數，超過回 413，直接回 200 + 結果。這其實是個純計算的 RPC，用 POST 是因為輸入太大放不進 query string。
• 大資料、非同步：client 先用 presigned URL 把兩個檔案上傳到物件儲存 → POST /v1/merge-jobs（帶兩個物件位置、合併參數、Idempotency-Key）→ 202 + Location → worker 依大小選擇記憶體合併或 Spark job → 結果寫回物件儲存 → GET /v1/merge-jobs/{id} 拿到結果下載連結，或 webhook 通知。

from typing import Literal
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI()

class MergeRequest(BaseModel):
    left: list[dict] = Field(max_length=50_000)
    right: list[dict] = Field(max_length=50_000)
    key: str = "id"
    how: Literal["inner", "left", "outer"] = "outer"
    conflict_policy: Literal["left", "right", "newest"] = "right"

class MergeResponse(BaseModel):
    data: list[dict]
    stats: dict

@app.post("/v1/merges", response_model=MergeResponse)
def merge(req: MergeRequest):           # 純 CPU 工作：用 def 讓 FastAPI 丟到 threadpool，不阻塞 event loop
    missing = sum(1 for r in req.left + req.right if req.key not in r)
    data = merge_records(req.left, req.right, req.key, req.how, req.conflict_policy)
    return MergeResponse(data=data, stats={"left": len(req.left), "right": len(req.right),
                                           "output": len(data), "skipped_missing_key": missing})

面試自測題