dramaling-vocab-learning/note/done/BACKEND_API_STRATEGY.md

🚀 後端API完整策略設計

📊 現狀分析

✅ 現有功能

• 基本詞卡CRUD操作
• 用戶收藏功能
• 基本資料結構完整

❌ 缺失功能

• 分頁功能 - 不支援page/limit參數
• 篩選功能 - difficultyLevel、partOfSpeech等參數無效
• 排序功能 - 不支援sortBy/sortOrder參數
• 搜尋功能 - 基本搜尋可能不完整
• 效能索引 - 缺少資料庫索引優化

---

🎯 完整API設計

核心端點：GET /api/flashcards

請求參數 (Query Parameters)

interface FlashcardQueryParams {
  // 搜尋和篩選
  search?: string;              // 全文搜尋 (詞彙、翻譯、定義)
  difficultyLevel?: string;     // CEFR等級 (A1, A2, B1, B2, C1, C2)
  partOfSpeech?: string;        // 詞性 (noun, verb, adjective, etc.)
  masteryLevel?: string;        // 掌握程度 (low: <60%, medium: 60-79%, high: 80%+)
  favoritesOnly?: boolean;      // 僅收藏詞卡

  // 時間範圍篩選
  createdAfter?: string;        // 創建時間起始 (ISO 8601)
  createdBefore?: string;       // 創建時間結束 (ISO 8601)
  reviewedAfter?: string;       // 最後複習時間起始
  reviewedBefore?: string;      // 最後複習時間結束

  // 複習統計篩選
  reviewCountMin?: number;      // 最少複習次數
  reviewCountMax?: number;      // 最多複習次數

  // 排序
  sortBy?: string;             // 排序字段 (createdAt, word, masteryLevel, difficultyLevel, timesReviewed)
  sortOrder?: 'asc' | 'desc';  // 排序方向

  // 分頁
  page?: number;               // 頁碼 (從1開始)
  limit?: number;              // 每頁數量 (預設20，最大100)

  // 其他
  includeMeta?: boolean;       // 是否包含元數據 (預設true)
}

標準回應格式

interface FlashcardQueryResponse {
  success: boolean;
  data: {
    flashcards: Flashcard[];
    pagination: {
      current_page: number;
      total_pages: number;
      total_count: number;
      page_size: number;
      has_next: boolean;
      has_prev: boolean;
    };
    filters_applied: {
      search?: string;
      difficulty_level?: string;
      part_of_speech?: string;
      mastery_level?: string;
      favorites_only?: boolean;
      // ... 其他應用的篩選
    };
    meta?: {
      query_time_ms: number;
      cache_hit: boolean;
    };
  };
  error?: string;
}

---

🏗️ 後端實作策略

階段一：基礎功能修復 (必需)

1.1 分頁功能實作

def get_flashcards():
    # 分頁參數
    page = int(request.args.get('page', 1))
    limit = min(int(request.args.get('limit', 20)), 100)  # 最大100
    offset = (page - 1) * limit

    # 構建基本查詢
    query = Flashcard.query.filter_by(user_id=current_user.id)

    # 應用篩選 (後續實作)
    query = apply_filters(query, request.args)

    # 應用排序 (後續實作)
    query = apply_sorting(query, request.args)

    # 計算總數 (在分頁之前)
    total_count = query.count()

    # 執行分頁查詢
    flashcards = query.offset(offset).limit(limit).all()

    # 計算分頁資訊
    total_pages = math.ceil(total_count / limit)

    return jsonify({
        'success': True,
        'data': {
            'flashcards': [card.to_dict() for card in flashcards],
            'pagination': {
                'current_page': page,
                'total_pages': total_pages,
                'total_count': total_count,
                'page_size': limit,
                'has_next': page < total_pages,
                'has_prev': page > 1
            }
        }
    })

1.2 篩選功能實作

def apply_filters(query, args):
    """應用所有篩選條件"""

    # 全文搜尋
    search = args.get('search')
    if search:
        search_pattern = f"%{search}%"
        query = query.filter(
            db.or_(
                Flashcard.word.ilike(search_pattern),
                Flashcard.translation.ilike(search_pattern),
                Flashcard.definition.ilike(search_pattern),
                Flashcard.example.ilike(search_pattern)
            )
        )

    # CEFR等級篩選
    difficulty_level = args.get('difficultyLevel')
    if difficulty_level:
        query = query.filter(Flashcard.difficulty_level == difficulty_level)

    # 詞性篩選
    part_of_speech = args.get('partOfSpeech')
    if part_of_speech:
        query = query.filter(Flashcard.part_of_speech == part_of_speech)

    # 掌握程度篩選
    mastery_level = args.get('masteryLevel')
    if mastery_level:
        if mastery_level == 'high':
            query = query.filter(Flashcard.mastery_level >= 80)
        elif mastery_level == 'medium':
            query = query.filter(
                Flashcard.mastery_level >= 60,
                Flashcard.mastery_level < 80
            )
        elif mastery_level == 'low':
            query = query.filter(Flashcard.mastery_level < 60)

    # 收藏篩選
    favorites_only = args.get('favoritesOnly', 'false').lower() == 'true'
    if favorites_only:
        query = query.filter(Flashcard.is_favorite == True)

    # 時間範圍篩選
    created_after = args.get('createdAfter')
    if created_after:
        query = query.filter(Flashcard.created_at >= created_after)

    created_before = args.get('createdBefore')
    if created_before:
        query = query.filter(Flashcard.created_at <= created_before)

    # 複習次數篩選
    review_count_min = args.get('reviewCountMin')
    if review_count_min:
        query = query.filter(Flashcard.times_reviewed >= int(review_count_min))

    review_count_max = args.get('reviewCountMax')
    if review_count_max:
        query = query.filter(Flashcard.times_reviewed <= int(review_count_max))

    return query

1.3 排序功能實作

def apply_sorting(query, args):
    """應用排序邏輯"""
    sort_by = args.get('sortBy', 'createdAt')
    sort_order = args.get('sortOrder', 'desc')

    # 排序字段映射
    sort_fields = {
        'createdAt': Flashcard.created_at,
        'word': Flashcard.word,
        'masteryLevel': Flashcard.mastery_level,
        'difficultyLevel': Flashcard.difficulty_level,
        'timesReviewed': Flashcard.times_reviewed
    }

    if sort_by not in sort_fields:
        sort_by = 'createdAt'  # 預設排序

    sort_field = sort_fields[sort_by]

    if sort_order.lower() == 'desc':
        query = query.order_by(sort_field.desc())
    else:
        query = query.order_by(sort_field.asc())

    # CEFR等級特殊排序
    if sort_by == 'difficultyLevel':
        # 需要自定義排序邏輯 A1 < A2 < B1 < B2 < C1 < C2
        level_order = case(
            (Flashcard.difficulty_level == 'A1', 1),
            (Flashcard.difficulty_level == 'A2', 2),
            (Flashcard.difficulty_level == 'B1', 3),
            (Flashcard.difficulty_level == 'B2', 4),
            (Flashcard.difficulty_level == 'C1', 5),
            (Flashcard.difficulty_level == 'C2', 6),
            else_=7
        )

        if sort_order.lower() == 'desc':
            query = query.order_by(level_order.desc())
        else:
            query = query.order_by(level_order.asc())

    return query

階段二：資料庫優化

2.1 索引建立

-- 基本篩選索引
CREATE INDEX idx_flashcards_user_difficulty ON flashcards(user_id, difficulty_level);
CREATE INDEX idx_flashcards_user_part_of_speech ON flashcards(user_id, part_of_speech);
CREATE INDEX idx_flashcards_user_mastery ON flashcards(user_id, mastery_level);
CREATE INDEX idx_flashcards_user_favorite ON flashcards(user_id, is_favorite);

-- 時間範圍索引
CREATE INDEX idx_flashcards_user_created_at ON flashcards(user_id, created_at);
CREATE INDEX idx_flashcards_user_times_reviewed ON flashcards(user_id, times_reviewed);

-- 複合索引 (重要查詢組合)
CREATE INDEX idx_flashcards_user_difficulty_mastery ON flashcards(user_id, difficulty_level, mastery_level);
CREATE INDEX idx_flashcards_user_favorite_created ON flashcards(user_id, is_favorite, created_at);

-- 全文搜尋索引 (PostgreSQL)
CREATE INDEX idx_flashcards_fulltext ON flashcards
USING gin(to_tsvector('english', word || ' ' || translation || ' ' || definition || ' ' || example));

2.2 查詢優化

# 使用 SQLAlchemy 查詢優化
def get_optimized_flashcards():
    # 使用子查詢優化計數
    subquery = apply_filters(
        Flashcard.query.filter_by(user_id=current_user.id),
        request.args
    ).subquery()

    # 總數查詢
    total_count = db.session.query(subquery).count()

    # 分頁查詢
    query = db.session.query(Flashcard).select_from(subquery)
    query = apply_sorting(query, request.args)

    flashcards = query.offset(offset).limit(limit).all()

    return flashcards, total_count

階段三：快取策略

3.1 Redis快取

import redis
import json
import hashlib

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def get_cache_key(user_id, params):
    """生成快取鍵"""
    cache_data = {
        'user_id': user_id,
        'params': dict(sorted(params.items()))
    }
    cache_string = json.dumps(cache_data, sort_keys=True)
    return f"flashcards:{hashlib.md5(cache_string.encode()).hexdigest()}"

def get_cached_flashcards(user_id, params):
    """從快取獲取結果"""
    cache_key = get_cache_key(user_id, params)
    cached_data = redis_client.get(cache_key)

    if cached_data:
        return json.loads(cached_data)

    return None

def cache_flashcards(user_id, params, data, ttl=300):
    """快取結果 (5分鐘TTL)"""
    cache_key = get_cache_key(user_id, params)
    redis_client.setex(cache_key, ttl, json.dumps(data))

3.2 快取失效策略

def invalidate_user_cache(user_id):
    """當用戶資料變更時清除快取"""
    pattern = f"flashcards:*user_id*{user_id}*"

    for key in redis_client.scan_iter(match=pattern):
        redis_client.delete(key)

# 在 CRUD 操作後調用
@app.after_request
def invalidate_cache_on_mutation(response):
    if request.method in ['POST', 'PUT', 'DELETE']:
        invalidate_user_cache(current_user.id)
    return response

---

🧪 API測試策略

測試案例設計

基本功能測試

# 1. 分頁測試
curl "http://localhost:5008/api/flashcards?page=1&limit=2"
curl "http://localhost:5008/api/flashcards?page=2&limit=2"

# 2. 篩選測試
curl "http://localhost:5008/api/flashcards?difficultyLevel=A2"
curl "http://localhost:5008/api/flashcards?partOfSpeech=noun"
curl "http://localhost:5008/api/flashcards?masteryLevel=low"

# 3. 排序測試
curl "http://localhost:5008/api/flashcards?sortBy=word&sortOrder=asc"
curl "http://localhost:5008/api/flashcards?sortBy=masteryLevel&sortOrder=desc"

# 4. 組合測試
curl "http://localhost:5008/api/flashcards?difficultyLevel=A2&sortBy=word&page=1&limit=5"

效能測試

# 負載測試
import time
import requests

def performance_test():
    start_time = time.time()

    response = requests.get('http://localhost:5008/api/flashcards', params={
        'search': 'test',
        'difficultyLevel': 'A2',
        'sortBy': 'createdAt',
        'page': 1,
        'limit': 20
    })

    end_time = time.time()
    response_time = (end_time - start_time) * 1000

    print(f"Response time: {response_time:.2f}ms")
    return response_time

# 目標：<300ms

---

📈 效能指標

成功標準

• 回應時間: < 300ms (95th percentile)
• 分頁查詢: < 200ms
• 搜尋查詢: < 500ms
• 快取命中率: > 60%
• 資料庫連接: < 100ms

監控指標

# API監控中間件
@app.before_request
def before_request():
    g.start_time = time.time()

@app.after_request
def after_request(response):
    response_time = (time.time() - g.start_time) * 1000

    # 記錄慢查詢
    if response_time > 500:
        logger.warning(f"Slow query: {request.url} - {response_time:.2f}ms")

    # 添加效能標頭
    response.headers['X-Response-Time'] = f"{response_time:.2f}ms"

    return response

---

🔄 實施計劃

第1週：基礎功能

• ✅ 實作分頁功能
• ✅ 實作基本篩選
• ✅ 實作排序功能
• ✅ API測試

第2週：優化與擴展

• ✅ 建立資料庫索引
• ✅ 實作進階篩選
• ✅ 效能優化
• ✅ 錯誤處理

第3週：快取與監控

• ✅ 實作Redis快取
• ✅ 效能監控
• ✅ 負載測試
• ✅ 文檔完善

這個完整的後端API策略確保：

• 🎯 功能完整 - 支援所有前端需求
• ⚡ 高效能 - 資料庫和快取優化
• 🔒 穩定性 - 完整的錯誤處理
• 📊 可監控 - 效能指標追蹤
• 🧪 可測試 - 完整的測試覆蓋

---

文檔版本: 1.0 最後更新: 2025-09-24