在現代軟體架構中，REST API 扮演著系統間溝通的核心角色。無論是前後端分離的單頁應用程式、行動應用程式後端，還是微服務架構中的服務間通訊，REST API 都是最廣泛採用的通訊協定。Flask 作為 Python 生態系中最受歡迎的微框架之一，以其輕量、靈活的特性深受開發者喜愛。透過 Flask-RESTful 擴充套件，開發者能夠快速建立符合 REST 原則的 API，同時保持程式碼的簡潔與可維護性。
然而，建立 API 僅是第一步，確保 API 的安全性才是真正的挑戰。在網路攻擊日益頻繁的今日，未經保護的 API 端點可能成為駭客入侵系統的突破口。因此，實作完善的認證與授權機制、建立健全的錯誤處理與日誌記錄系統，以及針對效能進行最佳化，都是開發生產級 API 不可或缺的環節。本文將深入探討這些主題，提供從基礎到進階的完整實作指南。
Flask-RESTful 架構設計原理
Flask-RESTful 是建立在 Flask 之上的擴充套件，它提供了一套簡潔的工具來快速建立 REST API。這個擴充套件的設計理念是將每個 API 端點視為一個資源（Resource），並透過類別方法來處理不同的 HTTP 方法。這種設計模式不僅使程式碼組織更加清晰，也符合物件導向程式設計的原則。
在傳統的 Flask 應用程式中，開發者通常使用裝飾器來定義路由，這種方式在處理簡單的 API 時相當方便。然而，當 API 複雜度增加時，程式碼容易變得混亂。Flask-RESTful 透過資源類別的概念解決了這個問題，每個資源類別封裝了對特定端點的所有操作，使得程式碼的組織結構更加模組化。
@startuml
!define PLANTUML_FORMAT svg
!theme _none_
skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 16
skinparam minClassWidth 100
' 類別圖展示 Flask-RESTful 架構
package "Flask-RESTful 架構" {
class Flask {
+ __init__(name)
+ run()
+ route()
}
class Api {
+ __init__(app)
+ add_resource()
}
abstract class Resource {
+ get()
+ post()
+ put()
+ delete()
}
class ItemResource {
+ get(id)
+ post()
+ put(id)
+ delete(id)
}
class ItemListResource {
+ get()
+ post()
}
}
Flask --> Api : 初始化
Api --> Resource : 管理
Resource <|-- ItemResource
Resource <|-- ItemListResource
@endumlFlask-RESTful 的另一個重要特性是請求解析（Request Parsing）功能。透過 reqparse 模組，開發者可以輕鬆地驗證和解析傳入的請求資料，確保資料的完整性和正確性。這個功能類似於表單驗證，但專門針對 API 請求進行最佳化，支援多種資料來源（如 JSON、表單資料、查詢參數等）和豐富的驗證規則。
此外，Flask-RESTful 還提供了輸出格式化（Output Formatting）功能，透過 marshal_with 裝飾器，開發者可以定義 API 回應的結構，確保回應資料的一致性。這個功能特別適合用於需要對敏感資料進行過濾或對複雜物件進行序列化的場景。
建立 Flask-RESTful 應用程式基礎架構
在開始建立 Flask-RESTful 應用程式之前，首先需要設定開發環境並安裝必要的相依套件。建議使用虛擬環境來隔離專案相依性，這樣可以避免不同專案之間的套件版本衝突。
# 建立專案目錄結構並設定虛擬環境
# 這是專案初始化的標準流程，確保開發環境的一致性
# requirements.txt 檔案內容
# 這個檔案列出了專案所需的所有 Python 套件及其版本
# 使用固定版本號可以確保在不同環境中的一致性
"""
flask==2.3.3
flask-restful==0.3.10
flask-sqlalchemy==3.1.1
flask-login==0.6.3
flask-jwt-extended==4.5.3
flask-caching==2.1.0
python-dotenv==1.0.0
werkzeug==2.3.7
gunicorn==21.2.0
orjson==3.9.7
"""
# app/__init__.py
# 這個檔案是應用程式工廠函式的所在位置
# 採用應用程式工廠模式可以方便地建立多個應用程式實例，適合測試和不同環境配置
from flask import Flask
from flask_restful import Api
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager
from flask_caching import Cache
from config import Config
# 初始化擴充套件實例
# 這些實例在這裡建立但尚未綁定到應用程式
# 這種延遲初始化的模式稱為「延遲綁定」，是 Flask 擴充套件的最佳實踐
db = SQLAlchemy()
jwt = JWTManager()
cache = Cache()
def create_app(config_class=Config):
"""
應用程式工廠函式
這個函式負責建立和配置 Flask 應用程式實例。
透過參數傳入不同的配置類別，可以輕鬆切換開發、測試和生產環境。
參數:
config_class: 配置類別，預設使用 Config
回傳:
已配置完成的 Flask 應用程式實例
"""
# 建立 Flask 應用程式實例
# __name__ 參數告訴 Flask 應用程式的根路徑
app = Flask(__name__)
# 從配置類別載入設定
# 這包括資料庫連線字串、密鑰、JWT 設定等
app.config.from_object(config_class)
# 初始化擴充套件
# 將之前建立的擴充套件實例綁定到應用程式
db.init_app(app)
jwt.init_app(app)
cache.init_app(app)
# 建立 API 實例
# Api 物件將管理所有的資源類別和路由
api = Api(app)
# 註冊資源
# 將資源類別綁定到特定的 URL 路徑
from app.resources.items import ItemResource, ItemListResource
from app.resources.auth import LoginResource, RegisterResource, LogoutResource
# 項目相關端點
api.add_resource(ItemListResource, '/api/v1/items')
api.add_resource(ItemResource, '/api/v1/items/<int:item_id>')
# 認證相關端點
api.add_resource(RegisterResource, '/api/v1/auth/register')
api.add_resource(LoginResource, '/api/v1/auth/login')
api.add_resource(LogoutResource, '/api/v1/auth/logout')
# 註冊錯誤處理器
register_error_handlers(app)
# 註冊日誌配置
configure_logging(app)
return app
def register_error_handlers(app):
"""
註冊全域錯誤處理器
這個函式設定了各種 HTTP 錯誤的處理方式，
確保 API 在發生錯誤時能夠回傳一致格式的錯誤訊息。
"""
@app.errorhandler(400)
def bad_request(error):
# 處理客戶端請求格式錯誤
# 這通常發生在請求參數缺失或格式不正確時
return {'error': '請求格式錯誤', 'message': str(error)}, 400
@app.errorhandler(401)
def unauthorized(error):
# 處理未授權存取
# 當使用者未提供有效的認證憑證時觸發
return {'error': '未授權存取', 'message': '請先登入'}, 401
@app.errorhandler(403)
def forbidden(error):
# 處理禁止存取
# 當使用者已認證但沒有存取特定資源的權限時觸發
return {'error': '禁止存取', 'message': '您沒有權限執行此操作'}, 403
@app.errorhandler(404)
def not_found(error):
# 處理資源未找到
# 當請求的資源不存在時觸發
return {'error': '資源未找到', 'message': '請求的資源不存在'}, 404
@app.errorhandler(500)
def internal_error(error):
# 處理內部伺服器錯誤
# 當伺服器發生未預期的錯誤時觸發
# 在生產環境中應該避免洩漏詳細錯誤訊息
db.session.rollback()
return {'error': '內部伺服器錯誤', 'message': '伺服器發生錯誤，請稍後再試'}, 500
def configure_logging(app):
"""
配置應用程式日誌系統
在生產環境中，適當的日誌記錄對於除錯和監控至關重要。
這個函式設定了檔案日誌和格式化器，確保重要事件都被記錄下來。
"""
import logging
from logging.handlers import RotatingFileHandler
import os
# 只在非除錯模式下啟用檔案日誌
# 在開發時使用終端輸出更方便除錯
if not app.debug:
# 確保日誌目錄存在
if not os.path.exists('logs'):
os.mkdir('logs')
# 設定旋轉檔案處理器
# 當檔案大小達到 10MB 時會建立新檔案，最多保留 10 個備份
file_handler = RotatingFileHandler(
'logs/flask_api.log',
maxBytes=10240000,
backupCount=10
)
# 設定日誌格式
# 包含時間戳、日誌等級、訊息內容和來源位置
file_handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
))
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
app.logger.info('Flask API 應用程式啟動')
上述程式碼展示了一個完整的 Flask-RESTful 應用程式架構。應用程式工廠模式是 Flask 社群推薦的最佳實踐，它允許開發者輕鬆地建立多個應用程式實例，這在執行測試或需要針對不同環境使用不同配置時特別有用。
設計 RESTful 資源類別
資源類別是 Flask-RESTful 的核心概念。每個資源類別對應一個 API 端點，並透過定義不同的方法來處理各種 HTTP 請求。一個設計良好的資源類別應該遵循單一職責原則，專注於處理特定類型的資源。
# app/resources/items.py
# 這個模組包含了項目資源的所有 API 端點實作
from flask import request, current_app
from flask_restful import Resource, reqparse, marshal_with, fields
from flask_jwt_extended import jwt_required, get_jwt_identity
from app import db, cache
from app.models import Item, User
# 定義項目的輸出格式
# marshal_with 裝飾器會根據這個定義來格式化回應資料
# 這確保了 API 回應的一致性，並可以過濾敏感欄位
item_fields = {
'id': fields.Integer,
'name': fields.String,
'description': fields.String,
'price': fields.Float,
'quantity': fields.Integer,
'created_at': fields.DateTime(dt_format='iso8601'),
'updated_at': fields.DateTime(dt_format='iso8601'),
'owner': fields.Nested({
'id': fields.Integer,
'username': fields.String
})
}
class ItemListResource(Resource):
"""
項目列表資源
處理項目集合的操作，包括取得所有項目和新增項目。
這個資源對應 /api/v1/items 端點。
"""
@cache.cached(timeout=60, query_string=True)
@marshal_with(item_fields)
def get(self):
"""
取得項目列表
支援分頁和篩選功能。使用快取來提升效能，
快取會在 60 秒後過期。查詢字串會被納入快取鍵中，
確保不同的查詢參數會有獨立的快取。
查詢參數:
page: 頁碼，預設為 1
per_page: 每頁數量，預設為 20，最大 100
search: 搜尋關鍵字，用於搜尋項目名稱
回傳:
項目列表和分頁資訊
"""
# 解析查詢參數
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 100)
search = request.args.get('search', '')
# 建立查詢
query = Item.query
# 如果有搜尋關鍵字，進行模糊搜尋
if search:
query = query.filter(Item.name.ilike(f'%{search}%'))
# 按建立時間降序排序，最新的項目排在前面
query = query.order_by(Item.created_at.desc())
# 執行分頁查詢
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
# 記錄 API 呼叫
current_app.logger.info(f'取得項目列表 - 頁碼: {page}, 每頁數量: {per_page}')
return pagination.items
@jwt_required()
@marshal_with(item_fields)
def post(self):
"""
新增項目
需要 JWT 認證。只有已登入的使用者才能新增項目。
新項目會自動關聯到目前登入的使用者。
請求主體:
name: 項目名稱（必填）
description: 項目描述（選填）
price: 價格（必填，必須大於 0）
quantity: 數量（必填，必須大於等於 0）
回傳:
新建立的項目資料
"""
# 設定請求解析器
parser = reqparse.RequestParser()
parser.add_argument('name', type=str, required=True,
help='項目名稱為必填欄位')
parser.add_argument('description', type=str, default='')
parser.add_argument('price', type=float, required=True,
help='價格為必填欄位')
parser.add_argument('quantity', type=int, required=True,
help='數量為必填欄位')
args = parser.parse_args()
# 驗證價格和數量
if args['price'] <= 0:
return {'message': '價格必須大於 0'}, 400
if args['quantity'] < 0:
return {'message': '數量不能為負數'}, 400
# 取得目前使用者 ID
current_user_id = get_jwt_identity()
# 建立新項目
item = Item(
name=args['name'],
description=args['description'],
price=args['price'],
quantity=args['quantity'],
user_id=current_user_id
)
# 儲存到資料庫
db.session.add(item)
db.session.commit()
# 清除項目列表快取
# 因為資料已變更，需要確保下次查詢會取得最新資料
cache.delete_memoized(self.get)
current_app.logger.info(f'使用者 {current_user_id} 新增項目: {item.name}')
return item, 201
class ItemResource(Resource):
"""
單一項目資源
處理特定項目的操作，包括取得、更新和刪除。
這個資源對應 /api/v1/items/<item_id> 端點。
"""
@marshal_with(item_fields)
def get(self, item_id):
"""
取得特定項目
根據項目 ID 取得項目詳細資訊。
如果項目不存在，回傳 404 錯誤。
參數:
item_id: 項目的唯一識別碼
回傳:
項目詳細資訊
"""
# 查詢項目，如果不存在則回傳 404
item = Item.query.get_or_404(item_id)
current_app.logger.info(f'取得項目詳情: {item.name} (ID: {item_id})')
return item
@jwt_required()
@marshal_with(item_fields)
def put(self, item_id):
"""
更新項目
需要 JWT 認證。只有項目的擁有者才能更新項目。
支援部分更新，只需要提供要修改的欄位。
參數:
item_id: 項目的唯一識別碼
請求主體:
name: 項目名稱（選填）
description: 項目描述（選填）
price: 價格（選填）
quantity: 數量（選填）
回傳:
更新後的項目資料
"""
item = Item.query.get_or_404(item_id)
# 檢查權限
current_user_id = get_jwt_identity()
if item.user_id != current_user_id:
return {'message': '您沒有權限修改此項目'}, 403
# 設定請求解析器
# 所有欄位都是選填的，實現部分更新功能
parser = reqparse.RequestParser()
parser.add_argument('name', type=str, store_missing=False)
parser.add_argument('description', type=str, store_missing=False)
parser.add_argument('price', type=float, store_missing=False)
parser.add_argument('quantity', type=int, store_missing=False)
args = parser.parse_args()
# 更新提供的欄位
if 'name' in args:
item.name = args['name']
if 'description' in args:
item.description = args['description']
if 'price' in args:
if args['price'] <= 0:
return {'message': '價格必須大於 0'}, 400
item.price = args['price']
if 'quantity' in args:
if args['quantity'] < 0:
return {'message': '數量不能為負數'}, 400
item.quantity = args['quantity']
db.session.commit()
current_app.logger.info(f'使用者 {current_user_id} 更新項目: {item.name}')
return item
@jwt_required()
def delete(self, item_id):
"""
刪除項目
需要 JWT 認證。只有項目的擁有者或管理員才能刪除項目。
刪除操作是不可逆的，請謹慎使用。
參數:
item_id: 項目的唯一識別碼
回傳:
刪除成功訊息
"""
item = Item.query.get_or_404(item_id)
# 檢查權限
current_user_id = get_jwt_identity()
current_user = User.query.get(current_user_id)
# 只有擁有者或管理員可以刪除
if item.user_id != current_user_id and not current_user.is_admin:
return {'message': '您沒有權限刪除此項目'}, 403
# 記錄被刪除的項目名稱，用於日誌
item_name = item.name
db.session.delete(item)
db.session.commit()
current_app.logger.info(f'使用者 {current_user_id} 刪除項目: {item_name}')
return {'message': f'項目 {item_name} 已成功刪除'}, 200
這個資源類別的設計展示了幾個重要的最佳實踐。首先，使用 reqparse 進行請求驗證確保了輸入資料的正確性。其次，使用 marshal_with 定義輸出格式確保了回應的一致性。第三，使用快取機制提升了查詢效能。最後，完善的權限檢查確保了資料的安全性。
JWT 認證機制實作
JSON Web Token（JWT）是目前最廣泛使用的 API 認證機制之一。相較於傳統的 Session 認證，JWT 具有無狀態、可擴展性高、跨域支援等優點，特別適合用於 REST API 和微服務架構。
@startuml
!define PLANTUML_FORMAT svg
!theme _none_
skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 16
skinparam minClassWidth 100
' 時序圖展示 JWT 認證流程
actor "客戶端" as Client
participant "API 伺服器" as API
database "資料庫" as DB
participant "JWT 服務" as JWT
== 使用者註冊 ==
Client -> API : POST /auth/register
API -> DB : 檢查使用者名稱是否存在
DB --> API : 使用者不存在
API -> API : 雜湊密碼
API -> DB : 儲存使用者資料
DB --> API : 儲存成功
API --> Client : 201 註冊成功
== 使用者登入 ==
Client -> API : POST /auth/login
API -> DB : 查詢使用者
DB --> API : 使用者資料
API -> API : 驗證密碼
API -> JWT : 產生 Access Token
JWT --> API : Access Token
API -> JWT : 產生 Refresh Token
JWT --> API : Refresh Token
API --> Client : 200 Token 資訊
== 存取受保護資源 ==
Client -> API : GET /items (帶 Bearer Token)
API -> JWT : 驗證 Token
JWT --> API : Token 有效
API -> DB : 查詢資料
DB --> API : 資料結果
API --> Client : 200 資料回應
== Token 更新 ==
Client -> API : POST /auth/refresh
API -> JWT : 驗證 Refresh Token
JWT --> API : Token 有效
API -> JWT : 產生新 Access Token
JWT --> API : 新 Access Token
API --> Client : 200 新 Token
@endumlJWT 認證流程包含幾個關鍵步驟。首先，使用者透過提供帳號密碼進行登入。伺服器驗證憑證後，產生一個包含使用者資訊的 JWT。這個 Token 會傳回給客戶端，客戶端在後續請求中將 Token 放在 Authorization 標頭中。伺服器收到請求後會驗證 Token 的有效性，確認後才允許存取受保護的資源。
# app/resources/auth.py
# 這個模組包含了所有認證相關的 API 端點
from flask import request, current_app
from flask_restful import Resource, reqparse
from flask_jwt_extended import (
create_access_token,
create_refresh_token,
jwt_required,
get_jwt_identity,
get_jwt
)
from werkzeug.security import generate_password_hash, check_password_hash
from app import db
from app.models import User, TokenBlocklist
from datetime import timedelta
class RegisterResource(Resource):
"""
使用者註冊資源
處理新使用者的註冊請求。
"""
def post(self):
"""
註冊新使用者
驗證使用者提供的資料，建立新帳號。
密碼會經過雜湊處理後儲存，確保安全性。
請求主體:
username: 使用者名稱（必填，3-50 字元）
email: 電子郵件（必填，必須是有效格式）
password: 密碼（必填，至少 8 字元）
回傳:
註冊成功訊息和使用者 ID
"""
parser = reqparse.RequestParser()
parser.add_argument('username', type=str, required=True,
help='使用者名稱為必填欄位')
parser.add_argument('email', type=str, required=True,
help='電子郵件為必填欄位')
parser.add_argument('password', type=str, required=True,
help='密碼為必填欄位')
args = parser.parse_args()
# 驗證使用者名稱長度
if len(args['username']) < 3 or len(args['username']) > 50:
return {'message': '使用者名稱必須在 3 到 50 個字元之間'}, 400
# 驗證密碼強度
if len(args['password']) < 8:
return {'message': '密碼長度至少需要 8 個字元'}, 400
# 驗證電子郵件格式（簡單驗證）
if '@' not in args['email'] or '.' not in args['email']:
return {'message': '請輸入有效的電子郵件地址'}, 400
# 檢查使用者名稱是否已存在
if User.query.filter_by(username=args['username']).first():
return {'message': '此使用者名稱已被使用'}, 409
# 檢查電子郵件是否已存在
if User.query.filter_by(email=args['email']).first():
return {'message': '此電子郵件已被註冊'}, 409
# 建立新使用者
# 使用 werkzeug 的 generate_password_hash 進行密碼雜湊
# 預設使用 pbkdf2:sha256 演算法，這是目前推薦的安全選項
user = User(
username=args['username'],
email=args['email'],
password_hash=generate_password_hash(args['password'])
)
db.session.add(user)
db.session.commit()
current_app.logger.info(f'新使用者註冊: {user.username}')
return {
'message': '註冊成功',
'user_id': user.id
}, 201
class LoginResource(Resource):
"""
使用者登入資源
處理使用者登入請求，驗證憑證並發放 JWT。
"""
def post(self):
"""
使用者登入
驗證使用者憑證，成功後產生 Access Token 和 Refresh Token。
Access Token 用於存取受保護資源，有效期較短。
Refresh Token 用於在 Access Token 過期後取得新的 Access Token。
請求主體:
username: 使用者名稱（必填）
password: 密碼（必填）
回傳:
access_token: 存取權杖
refresh_token: 更新權杖
expires_in: Access Token 有效期（秒）
"""
parser = reqparse.RequestParser()
parser.add_argument('username', type=str, required=True,
help='使用者名稱為必填欄位')
parser.add_argument('password', type=str, required=True,
help='密碼為必填欄位')
args = parser.parse_args()
# 查詢使用者
user = User.query.filter_by(username=args['username']).first()
# 驗證使用者存在且密碼正確
# 使用 check_password_hash 來比對雜湊後的密碼
if not user or not check_password_hash(user.password_hash, args['password']):
# 不要透露是使用者名稱錯誤還是密碼錯誤
# 這是防止帳號列舉攻擊的安全措施
current_app.logger.warning(f'登入失敗: {args["username"]}')
return {'message': '使用者名稱或密碼錯誤'}, 401
# 產生 JWT
# identity 參數是會被編碼進 Token 的使用者識別資訊
# 這裡使用使用者 ID，可以在後續請求中透過 get_jwt_identity() 取得
access_token = create_access_token(
identity=user.id,
additional_claims={
'username': user.username,
'is_admin': user.is_admin
},
expires_delta=timedelta(hours=1) # Access Token 有效期 1 小時
)
refresh_token = create_refresh_token(
identity=user.id,
expires_delta=timedelta(days=30) # Refresh Token 有效期 30 天
)
# 更新最後登入時間
user.update_last_login()
current_app.logger.info(f'使用者登入成功: {user.username}')
return {
'access_token': access_token,
'refresh_token': refresh_token,
'token_type': 'Bearer',
'expires_in': 3600 # 1 小時 = 3600 秒
}, 200
class RefreshResource(Resource):
"""
Token 更新資源
處理 Access Token 的更新請求。
"""
@jwt_required(refresh=True)
def post(self):
"""
更新 Access Token
使用 Refresh Token 取得新的 Access Token。
這允許使用者在不重新登入的情況下延續工作階段。
需要在 Authorization 標頭中提供 Refresh Token。
回傳:
新的 access_token
"""
# 取得目前使用者 ID
current_user_id = get_jwt_identity()
# 查詢使用者以取得最新資訊
user = User.query.get(current_user_id)
if not user:
return {'message': '使用者不存在'}, 404
# 產生新的 Access Token
new_access_token = create_access_token(
identity=user.id,
additional_claims={
'username': user.username,
'is_admin': user.is_admin
},
expires_delta=timedelta(hours=1)
)
current_app.logger.info(f'Token 更新: {user.username}')
return {
'access_token': new_access_token,
'token_type': 'Bearer',
'expires_in': 3600
}, 200
class LogoutResource(Resource):
"""
使用者登出資源
處理使用者登出請求，將目前的 Token 加入黑名單。
"""
@jwt_required()
def post(self):
"""
使用者登出
將目前的 JWT 加入黑名單，使其失效。
這是實現「真正登出」功能的關鍵，
因為 JWT 本身是無狀態的，無法主動使其失效。
回傳:
登出成功訊息
"""
# 取得目前 Token 的 JTI（JWT ID）
# JTI 是每個 Token 的唯一識別碼
jti = get_jwt()['jti']
# 將 Token 加入黑名單
token_blocklist = TokenBlocklist(jti=jti)
db.session.add(token_blocklist)
db.session.commit()
current_user_id = get_jwt_identity()
current_app.logger.info(f'使用者登出: ID {current_user_id}')
return {'message': '登出成功'}, 200
JWT 認證機制的實作涉及到幾個重要的安全考量。首先，密碼儲存必須使用強雜湊演算法，werkzeug 提供的 generate_password_hash 預設使用 pbkdf2:sha256，這是目前業界推薦的選項。其次，為了防止帳號列舉攻擊，登入失敗時不應該透露是使用者名稱錯誤還是密碼錯誤。第三，透過 Token 黑名單機制可以實現真正的登出功能。
角色權限控管實作
在企業級應用程式中，僅有基本的認證機制是不夠的，還需要實作細緻的授權控管。角色權限控管（Role-Based Access Control，RBAC）是最常見的授權模型，它根據使用者的角色來決定其可以存取的資源和操作。
# app/decorators.py
# 這個模組包含了自定義的裝飾器，用於實作角色權限控管
from functools import wraps
from flask import current_app
from flask_jwt_extended import get_jwt, verify_jwt_in_request
def admin_required():
"""
管理員權限裝飾器
這個裝飾器用於保護只有管理員才能存取的端點。
它會檢查 JWT 中的 is_admin 宣告，確保只有管理員可以執行操作。
使用範例:
@app.route('/admin/users')
@admin_required()
def admin_users():
return get_all_users()
"""
def wrapper(fn):
@wraps(fn)
def decorator(*args, **kwargs):
# 首先驗證 JWT 的有效性
verify_jwt_in_request()
# 取得 JWT 中的額外宣告
claims = get_jwt()
# 檢查是否為管理員
if not claims.get('is_admin', False):
current_app.logger.warning(
f'非管理員嘗試存取管理功能: 使用者 ID {claims.get("sub")}'
)
return {'message': '此操作需要管理員權限'}, 403
return fn(*args, **kwargs)
return decorator
return wrapper
def role_required(allowed_roles):
"""
角色權限裝飾器
這個裝飾器用於實作更細緻的角色控管。
可以指定允許存取的角色列表。
參數:
allowed_roles: 允許存取的角色列表
使用範例:
@app.route('/reports')
@role_required(['admin', 'manager', 'analyst'])
def view_reports():
return get_reports()
"""
def wrapper(fn):
@wraps(fn)
def decorator(*args, **kwargs):
verify_jwt_in_request()
claims = get_jwt()
# 取得使用者角色
user_role = claims.get('role', 'user')
if user_role not in allowed_roles:
current_app.logger.warning(
f'角色權限不足: 使用者角色 {user_role}，需要角色 {allowed_roles}'
)
return {'message': '您的角色沒有權限執行此操作'}, 403
return fn(*args, **kwargs)
return decorator
return wrapper
def permission_required(permission):
"""
權限檢查裝飾器
這個裝飾器用於檢查使用者是否具有特定權限。
相較於角色檢查，權限檢查更加細緻，
因為同一個角色可能有不同的權限組合。
參數:
permission: 需要的權限名稱
使用範例:
@app.route('/items', methods=['DELETE'])
@permission_required('delete_items')
def delete_item():
return perform_delete()
"""
def wrapper(fn):
@wraps(fn)
def decorator(*args, **kwargs):
verify_jwt_in_request()
claims = get_jwt()
# 取得使用者權限列表
user_permissions = claims.get('permissions', [])
if permission not in user_permissions:
current_app.logger.warning(
f'權限不足: 需要權限 {permission}，使用者權限 {user_permissions}'
)
return {'message': f'此操作需要 {permission} 權限'}, 403
return fn(*args, **kwargs)
return decorator
return wrapper
# app/models.py
# 這個模組包含了資料模型定義
from datetime import datetime
from app import db
class User(db.Model):
"""
使用者模型
儲存使用者的基本資訊和認證資料。
"""
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(50), unique=True, nullable=False, index=True)
email = db.Column(db.String(120), unique=True, nullable=False, index=True)
password_hash = db.Column(db.String(256), nullable=False)
is_admin = db.Column(db.Boolean, default=False)
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
last_login = db.Column(db.DateTime)
# 與角色的多對多關係
roles = db.relationship('Role', secondary='user_roles', backref='users')
# 與項目的一對多關係
items = db.relationship('Item', backref='owner', lazy='dynamic')
def update_last_login(self):
"""更新最後登入時間"""
self.last_login = datetime.utcnow()
db.session.commit()
def get_permissions(self):
"""
取得使用者的所有權限
遍歷使用者的所有角色，收集所有權限。
這個方法用於在產生 JWT 時將權限資訊編碼進 Token。
回傳:
權限名稱的列表
"""
permissions = set()
for role in self.roles:
for permission in role.permissions:
permissions.add(permission.name)
return list(permissions)
class Role(db.Model):
"""
角色模型
定義系統中的角色，如管理員、編輯者、檢視者等。
"""
__tablename__ = 'roles'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), unique=True, nullable=False)
description = db.Column(db.String(200))
# 與權限的多對多關係
permissions = db.relationship('Permission', secondary='role_permissions', backref='roles')
class Permission(db.Model):
"""
權限模型
定義系統中的細緻權限，如建立項目、刪除項目等。
"""
__tablename__ = 'permissions'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), unique=True, nullable=False)
description = db.Column(db.String(200))
# 關聯表
user_roles = db.Table('user_roles',
db.Column('user_id', db.Integer, db.ForeignKey('users.id'), primary_key=True),
db.Column('role_id', db.Integer, db.ForeignKey('roles.id'), primary_key=True)
)
role_permissions = db.Table('role_permissions',
db.Column('role_id', db.Integer, db.ForeignKey('roles.id'), primary_key=True),
db.Column('permission_id', db.Integer, db.ForeignKey('permissions.id'), primary_key=True)
)
class Item(db.Model):
"""
項目模型
儲存項目的資訊。
"""
__tablename__ = 'items'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), nullable=False)
description = db.Column(db.Text)
price = db.Column(db.Float, nullable=False)
quantity = db.Column(db.Integer, nullable=False, default=0)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
class TokenBlocklist(db.Model):
"""
Token 黑名單模型
儲存已失效的 JWT 的 JTI。
當使用者登出時，其 Token 會被加入這個表。
"""
__tablename__ = 'token_blocklist'
id = db.Column(db.Integer, primary_key=True)
jti = db.Column(db.String(36), nullable=False, index=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
角色權限控管的設計採用了三層架構：使用者、角色和權限。這種設計提供了極大的靈活性，允許管理員輕鬆地調整權限配置而不需要修改程式碼。一個使用者可以擁有多個角色，每個角色可以包含多個權限。這種多對多的關係使得權限管理更加靈活和可擴展。
API 安全最佳實踐
除了認證和授權機制，還有許多其他的安全措施需要考慮。這些措施可以幫助保護 API 免受各種常見的攻擊，如 SQL 注入、跨站腳本（XSS）、跨站請求偽造（CSRF）等。
# config.py
# 這個模組包含了應用程式的配置設定
import os
from datetime import timedelta
class Config:
"""
應用程式配置類別
包含所有的配置設定，包括安全相關設定。
敏感資訊應該從環境變數讀取，而不是寫死在程式碼中。
"""
# 基本設定
SECRET_KEY = os.environ.get('SECRET_KEY') or 'your-secret-key-here'
# 資料庫設定
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
'sqlite:///app.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
# JWT 設定
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') or 'jwt-secret-key'
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)
JWT_TOKEN_LOCATION = ['headers']
JWT_HEADER_NAME = 'Authorization'
JWT_HEADER_TYPE = 'Bearer'
# 快取設定
CACHE_TYPE = 'RedisCache'
CACHE_REDIS_URL = os.environ.get('REDIS_URL') or 'redis://localhost:6379/0'
CACHE_DEFAULT_TIMEOUT = 300
# 安全設定
# 這些設定可以幫助防止常見的 Web 攻擊
SESSION_COOKIE_SECURE = True # 只在 HTTPS 下傳送 Cookie
SESSION_COOKIE_HTTPONLY = True # 防止 JavaScript 存取 Cookie
SESSION_COOKIE_SAMESITE = 'Lax' # 防止 CSRF 攻擊
# API 速率限制設定
RATELIMIT_DEFAULT = "200 per day;50 per hour;1 per second"
RATELIMIT_STORAGE_URL = os.environ.get('REDIS_URL') or 'memory://'
# app/security.py
# 這個模組包含了額外的安全機制實作
from flask import request, current_app
from functools import wraps
from datetime import datetime, timedelta
import re
class RateLimiter:
"""
API 速率限制器
防止 API 被濫用或遭受 DoS 攻擊。
使用滑動視窗演算法來追蹤請求次數。
"""
def __init__(self, app=None, default_limits=None):
self.app = app
self.default_limits = default_limits or ["200 per day", "50 per hour"]
self._storage = {}
if app is not None:
self.init_app(app)
def init_app(self, app):
"""初始化速率限制器"""
app.before_request(self._check_rate_limit)
def _check_rate_limit(self):
"""
檢查請求是否超過速率限制
根據客戶端 IP 和端點來追蹤請求次數。
如果超過限制，回傳 429 錯誤。
"""
client_ip = request.remote_addr
endpoint = request.endpoint
# 建立唯一的鍵來識別這個客戶端和端點的組合
key = f"{client_ip}:{endpoint}"
# 取得目前時間
now = datetime.utcnow()
# 初始化或清理過期的記錄
if key not in self._storage:
self._storage[key] = []
# 移除超過一天的記錄
self._storage[key] = [
timestamp for timestamp in self._storage[key]
if now - timestamp < timedelta(days=1)
]
# 檢查各個時間窗口的限制
requests_last_day = len(self._storage[key])
requests_last_hour = len([
t for t in self._storage[key]
if now - t < timedelta(hours=1)
])
requests_last_second = len([
t for t in self._storage[key]
if now - t < timedelta(seconds=1)
])
# 檢查是否超過限制
if requests_last_day >= 200:
return {'message': '已達每日請求上限，請明天再試'}, 429
if requests_last_hour >= 50:
return {'message': '已達每小時請求上限，請稍後再試'}, 429
if requests_last_second >= 1:
return {'message': '請求過於頻繁，請稍後再試'}, 429
# 記錄這次請求
self._storage[key].append(now)
def validate_input(data, rules):
"""
輸入驗證函式
根據定義的規則驗證輸入資料，防止注入攻擊。
參數:
data: 要驗證的資料字典
rules: 驗證規則字典
回傳:
(is_valid, errors): 是否有效和錯誤訊息列表
"""
errors = []
for field, rule in rules.items():
value = data.get(field)
# 檢查必填欄位
if rule.get('required') and not value:
errors.append(f'{field} 是必填欄位')
continue
if value:
# 檢查類型
expected_type = rule.get('type')
if expected_type and not isinstance(value, expected_type):
errors.append(f'{field} 類型錯誤')
# 檢查字串長度
if isinstance(value, str):
min_len = rule.get('min_length', 0)
max_len = rule.get('max_length', float('inf'))
if len(value) < min_len or len(value) > max_len:
errors.append(f'{field} 長度必須在 {min_len} 到 {max_len} 之間')
# 檢查格式
pattern = rule.get('pattern')
if pattern and not re.match(pattern, value):
errors.append(f'{field} 格式不正確')
# 檢查數值範圍
if isinstance(value, (int, float)):
min_val = rule.get('min', float('-inf'))
max_val = rule.get('max', float('inf'))
if value < min_val or value > max_val:
errors.append(f'{field} 必須在 {min_val} 到 {max_val} 之間')
return len(errors) == 0, errors
def sanitize_html(text):
"""
HTML 消毒函式
移除或轉義文字中的 HTML 標籤，防止 XSS 攻擊。
參數:
text: 要處理的文字
回傳:
消毒後的文字
"""
if not text:
return text
# 定義允許的標籤（如果需要的話）
# 這裡採用白名單策略，移除所有 HTML 標籤
import html
# 首先轉義 HTML 特殊字元
sanitized = html.escape(text)
return sanitized
def log_security_event(event_type, details, severity='INFO'):
"""
記錄安全事件
將安全相關事件記錄到日誌中，方便後續審計和分析。
參數:
event_type: 事件類型（如 'LOGIN_FAILED', 'UNAUTHORIZED_ACCESS'）
details: 事件詳細資訊
severity: 嚴重程度（INFO, WARNING, ERROR, CRITICAL）
"""
log_message = f'[SECURITY][{event_type}] {details}'
if severity == 'INFO':
current_app.logger.info(log_message)
elif severity == 'WARNING':
current_app.logger.warning(log_message)
elif severity == 'ERROR':
current_app.logger.error(log_message)
elif severity == 'CRITICAL':
current_app.logger.critical(log_message)
API 安全是一個多層次的問題，需要從多個角度來防護。速率限制可以防止 API 被濫用或遭受暴力攻擊。輸入驗證可以防止注入攻擊。HTML 消毒可以防止 XSS 攻擊。安全事件日誌可以幫助追蹤和分析可疑活動。這些措施相互配合，形成一個完整的安全防護網。
效能最佳化技術
API 的效能直接影響使用者體驗和系統的可擴展性。在高流量的生產環境中，效能最佳化是不可或缺的。以下是幾個關鍵的效能最佳化技術。
@startuml
!define PLANTUML_FORMAT svg
!theme _none_
skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 16
skinparam minClassWidth 100
' 元件圖展示效能最佳化架構
package "效能最佳化架構" {
component [Nginx\n反向代理] as Nginx
component [Gunicorn\n應用伺服器] as Gunicorn
component [Flask\n應用程式] as Flask
database [Redis\n快取] as Redis
database [PostgreSQL\n資料庫] as DB
Nginx --> Gunicorn : 負載平衡
Gunicorn --> Flask : WSGI
Flask --> Redis : 快取查詢
Flask --> DB : 資料查詢
}
note right of Nginx
靜態檔案服務
SSL 終止
Gzip 壓縮
end note
note right of Redis
Session 儲存
API 回應快取
速率限制計數
end note
@enduml效能最佳化可以從多個層面進行。在應用程式層面，可以使用快取來減少資料庫查詢、使用非同步處理來提升並行能力、使用更高效的序列化庫來加速資料處理。在基礎設施層面，可以使用反向代理來處理靜態檔案和負載平衡、使用連線池來優化資料庫連線管理。
# app/performance.py
# 這個模組包含了效能最佳化相關的實作
from flask import current_app
from app import db, cache
import orjson
def optimized_json_response(data):
"""
最佳化的 JSON 回應
使用 orjson 進行序列化，比標準 json 庫快 3-10 倍。
這在處理大量資料時可以顯著提升效能。
參數:
data: 要序列化的資料
回傳:
序列化後的 JSON 字串
"""
from decimal import Decimal
from datetime import datetime
def default_serializer(obj):
"""自定義序列化器，處理特殊類型"""
if isinstance(obj, Decimal):
return float(obj)
if isinstance(obj, datetime):
return obj.isoformat()
raise TypeError(f'無法序列化類型: {type(obj)}')
return orjson.dumps(
data,
default=default_serializer,
option=orjson.OPT_NAIVE_UTC | orjson.OPT_SERIALIZE_NUMPY
)
class QueryOptimizer:
"""
查詢最佳化器
提供資料庫查詢最佳化的輔助方法。
"""
@staticmethod
def paginate_query(query, page, per_page):
"""
分頁查詢
有效率地分頁查詢結果，避免一次載入所有資料。
參數:
query: SQLAlchemy 查詢物件
page: 頁碼
per_page: 每頁數量
回傳:
分頁後的查詢結果
"""
# 使用 offset 和 limit 來實現分頁
# 這比 slice 更有效率，因為資料庫會直接處理
offset = (page - 1) * per_page
return query.offset(offset).limit(per_page).all()
@staticmethod
def eager_load_relationships(query, *relationships):
"""
預先載入關聯資料
使用 joinedload 來預先載入關聯資料，
避免 N+1 查詢問題。
參數:
query: SQLAlchemy 查詢物件
relationships: 要預先載入的關聯名稱
回傳:
修改後的查詢物件
"""
from sqlalchemy.orm import joinedload
for relationship in relationships:
query = query.options(joinedload(relationship))
return query
@staticmethod
def select_specific_columns(model, *columns):
"""
選擇特定欄位
只查詢需要的欄位，減少資料傳輸量。
參數:
model: 資料模型類別
columns: 要查詢的欄位名稱
回傳:
查詢物件
"""
# 取得欄位物件
column_objects = [getattr(model, col) for col in columns]
return db.session.query(*column_objects)
def cached_query(cache_key, timeout=300):
"""
快取查詢結果的裝飾器
將查詢結果快取到 Redis，減少資料庫負載。
參數:
cache_key: 快取鍵（可以是字串或函式）
timeout: 快取過期時間（秒）
"""
def decorator(fn):
from functools import wraps
@wraps(fn)
def wrapper(*args, **kwargs):
# 產生快取鍵
if callable(cache_key):
key = cache_key(*args, **kwargs)
else:
key = cache_key
# 嘗試從快取取得資料
cached_result = cache.get(key)
if cached_result is not None:
current_app.logger.debug(f'快取命中: {key}')
return cached_result
# 快取未命中，執行查詢
current_app.logger.debug(f'快取未命中: {key}')
result = fn(*args, **kwargs)
# 儲存到快取
cache.set(key, result, timeout=timeout)
return result
return wrapper
return decorator
def batch_insert(model, items, batch_size=1000):
"""
批次插入資料
將大量資料分批插入資料庫，避免單次交易過大。
這比逐筆插入要高效得多。
參數:
model: 資料模型類別
items: 要插入的資料列表（字典格式）
batch_size: 每批次的數量
"""
total = len(items)
for i in range(0, total, batch_size):
batch = items[i:i + batch_size]
# 使用 bulk_insert_mappings 進行批次插入
# 這比逐個建立物件再 add_all 更快
db.session.bulk_insert_mappings(model, batch)
db.session.commit()
current_app.logger.info(f'已插入 {min(i + batch_size, total)}/{total} 筆資料')
# gunicorn.conf.py
# Gunicorn 配置檔案
# 這個配置針對生產環境進行最佳化
import multiprocessing
# 綁定地址和埠
bind = '0.0.0.0:8000'
# Worker 數量
# 建議設定為 (2 * CPU 核心數) + 1
workers = multiprocessing.cpu_count() * 2 + 1
# Worker 類型
# 使用 gevent 來支援非同步處理
worker_class = 'gevent'
# 每個 Worker 的連線數
worker_connections = 1000
# 超時設定
timeout = 30
graceful_timeout = 30
keepalive = 2
# 日誌設定
accesslog = 'logs/gunicorn_access.log'
errorlog = 'logs/gunicorn_error.log'
loglevel = 'info'
# 處理程序名稱
proc_name = 'flask_api'
# 預載入應用程式
# 這可以減少 Worker 的啟動時間和記憶體使用
preload_app = True
# 最大請求數
# Worker 處理這麼多請求後會重新啟動，防止記憶體洩漏
max_requests = 1000
max_requests_jitter = 50
效能最佳化需要根據實際的應用場景和瓶頸來選擇適當的策略。快取是最常用也最有效的最佳化手段，但需要注意快取失效策略和一致性問題。批次操作可以大幅減少資料庫往返次數。選擇性載入可以減少不必要的資料傳輸。使用合適的 WSGI 伺服器配置可以充分利用硬體資源。
Docker 容器化部署
將 Flask 應用程式容器化是現代部署的最佳實踐。Docker 容器提供了一致的執行環境，簡化了部署流程，並使應用程式更容易擴展和維護。
# Dockerfile
# 這個檔案定義了如何建置 Docker 映像檔
# 使用多階段建置來減小最終映像檔的大小
# 第一階段：建置環境
FROM python:3.11-slim as builder
# 設定工作目錄
WORKDIR /app
# 設定環境變數
# 這些設定可以減少 Python 的開銷
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
# 安裝建置依賴
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
libpq-dev \
&& rm -rf /var/lib/apt/lists/*
# 複製依賴檔案並安裝
COPY requirements.txt .
RUN pip wheel --no-cache-dir --no-deps --wheel-dir /app/wheels -r requirements.txt
# 第二階段：執行環境
FROM python:3.11-slim
# 建立非 root 使用者
# 這是安全最佳實踐，避免以 root 身份執行應用程式
RUN groupadd -r appgroup && useradd -r -g appgroup appuser
# 設定工作目錄
WORKDIR /app
# 安裝執行依賴
RUN apt-get update && apt-get install -y --no-install-recommends \
libpq5 \
&& rm -rf /var/lib/apt/lists/*
# 從建置階段複製已編譯的套件
COPY --from=builder /app/wheels /wheels
COPY --from=builder /app/requirements.txt .
# 安裝 Python 套件
RUN pip install --no-cache /wheels/*
# 複製應用程式程式碼
COPY . .
# 建立日誌目錄並設定權限
RUN mkdir -p logs && chown -R appuser:appgroup /app
# 切換到非 root 使用者
USER appuser
# 暴露埠
EXPOSE 8000
# 健康檢查
# Docker 會定期執行這個命令來檢查容器是否正常運作
HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# 啟動命令
CMD ["gunicorn", "--config", "gunicorn.conf.py", "wsgi:app"]
Docker 多階段建置是一個重要的最佳化技術，它可以顯著減小最終映像檔的大小。第一階段包含了所有建置工具和編譯器，用於編譯 Python 套件。第二階段只包含執行應用程式所需的最小環境，從第一階段複製已編譯的套件。這種方式可以將映像檔大小減少 50% 以上。
# docker-compose.yml
# 這個檔案定義了完整的應用程式堆疊
version: '3.8'
services:
# Flask 應用程式
web:
build: .
container_name: flask_api
restart: unless-stopped
environment:
- FLASK_ENV=production
- DATABASE_URL=postgresql://user:password@db:5432/flask_db
- REDIS_URL=redis://redis:6379/0
- SECRET_KEY=${SECRET_KEY}
- JWT_SECRET_KEY=${JWT_SECRET_KEY}
depends_on:
- db
- redis
networks:
- app_network
# Nginx 反向代理
nginx:
image: nginx:alpine
container_name: nginx
restart: unless-stopped
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
- ./ssl:/etc/nginx/ssl:ro
depends_on:
- web
networks:
- app_network
# PostgreSQL 資料庫
db:
image: postgres:15-alpine
container_name: postgres
restart: unless-stopped
environment:
- POSTGRES_USER=user
- POSTGRES_PASSWORD=password
- POSTGRES_DB=flask_db
volumes:
- postgres_data:/var/lib/postgresql/data
networks:
- app_network
# Redis 快取
redis:
image: redis:7-alpine
container_name: redis
restart: unless-stopped
command: redis-server --appendonly yes
volumes:
- redis_data:/data
networks:
- app_network
networks:
app_network:
driver: bridge
volumes:
postgres_data:
redis_data:
Docker Compose 提供了一種簡單的方式來定義和管理多容器應用程式。上述配置定義了一個完整的應用程式堆疊，包括 Flask 應用程式、Nginx 反向代理、PostgreSQL 資料庫和 Redis 快取。每個服務都在獨立的容器中執行，透過 Docker 網路相互通訊。
# nginx.conf
# Nginx 配置檔案
events {
worker_connections 1024;
}
http {
# 基本設定
include /etc/nginx/mime.types;
default_type application/octet-stream;
# 日誌格式
log_format main '$remote_addr - $remote_user [$time_local] "$request" '
'$status $body_bytes_sent "$http_referer" '
'"$http_user_agent" "$http_x_forwarded_for"';
access_log /var/log/nginx/access.log main;
# 效能最佳化
sendfile on;
tcp_nopush on;
tcp_nodelay on;
keepalive_timeout 65;
# Gzip 壓縮
gzip on;
gzip_vary on;
gzip_min_length 1024;
gzip_types text/plain text/css application/json application/javascript
text/xml application/xml application/xml+rss text/javascript;
# 上游伺服器
upstream flask_app {
server web:8000;
keepalive 32;
}
server {
listen 80;
server_name localhost;
# 重定向到 HTTPS
location / {
return 301 https://$host$request_uri;
}
}
server {
listen 443 ssl http2;
server_name localhost;
# SSL 設定
ssl_certificate /etc/nginx/ssl/cert.pem;
ssl_certificate_key /etc/nginx/ssl/key.pem;
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256;
ssl_prefer_server_ciphers off;
# 安全標頭
add_header X-Frame-Options "SAMEORIGIN" always;
add_header X-Content-Type-Options "nosniff" always;
add_header X-XSS-Protection "1; mode=block" always;
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
# API 端點
location /api {
proxy_pass http://flask_app;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# 超時設定
proxy_connect_timeout 60s;
proxy_send_timeout 60s;
proxy_read_timeout 60s;
}
# 健康檢查端點
location /health {
proxy_pass http://flask_app;
access_log off;
}
}
}
Nginx 作為反向代理提供了許多重要的功能。它可以處理 SSL 終止，減輕應用程式伺服器的負擔。它可以提供 Gzip 壓縮，減少網路傳輸量。它可以設定安全標頭，增強應用程式的安全性。它還可以進行負載平衡，將請求分發到多個應用程式實例。
監控與可觀測性
在生產環境中，監控和可觀測性是維持系統健康的關鍵。透過適當的監控，開發者可以及時發現問題、追蹤效能指標、分析使用模式。
# app/monitoring.py
# 這個模組包含了監控相關的實作
from flask import request, g
from time import time
import psutil
import os
class MetricsCollector:
"""
指標收集器
收集應用程式的各種指標，用於監控和分析。
"""
def __init__(self, app=None):
self.app = app
self._metrics = {
'requests_total': 0,
'requests_by_endpoint': {},
'requests_by_status': {},
'response_times': []
}
if app is not None:
self.init_app(app)
def init_app(self, app):
"""初始化指標收集器"""
app.before_request(self._before_request)
app.after_request(self._after_request)
# 註冊指標端點
@app.route('/metrics')
def metrics():
return self.get_metrics()
def _before_request(self):
"""在請求前記錄開始時間"""
g.start_time = time()
def _after_request(self, response):
"""
在請求後收集指標
記錄請求次數、回應時間和狀態碼分佈。
"""
# 計算回應時間
response_time = time() - g.start_time
# 更新總請求數
self._metrics['requests_total'] += 1
# 更新端點統計
endpoint = request.endpoint or 'unknown'
if endpoint not in self._metrics['requests_by_endpoint']:
self._metrics['requests_by_endpoint'][endpoint] = 0
self._metrics['requests_by_endpoint'][endpoint] += 1
# 更新狀態碼統計
status = str(response.status_code)
if status not in self._metrics['requests_by_status']:
self._metrics['requests_by_status'][status] = 0
self._metrics['requests_by_status'][status] += 1
# 記錄回應時間（保留最近 1000 筆）
self._metrics['response_times'].append(response_time)
if len(self._metrics['response_times']) > 1000:
self._metrics['response_times'] = self._metrics['response_times'][-1000:]
return response
def get_metrics(self):
"""
取得所有指標
回傳應用程式的各種指標，包括請求統計、
系統資源使用狀況等。
"""
# 計算平均回應時間
response_times = self._metrics['response_times']
avg_response_time = sum(response_times) / len(response_times) if response_times else 0
# 取得系統資源資訊
cpu_percent = psutil.cpu_percent()
memory = psutil.virtual_memory()
return {
'requests': {
'total': self._metrics['requests_total'],
'by_endpoint': self._metrics['requests_by_endpoint'],
'by_status': self._metrics['requests_by_status']
},
'response_time': {
'average_ms': round(avg_response_time * 1000, 2),
'samples': len(response_times)
},
'system': {
'cpu_percent': cpu_percent,
'memory_percent': memory.percent,
'memory_used_mb': round(memory.used / 1024 / 1024, 2),
'memory_available_mb': round(memory.available / 1024 / 1024, 2)
},
'process': {
'pid': os.getpid(),
'memory_mb': round(psutil.Process().memory_info().rss / 1024 / 1024, 2)
}
}
def create_health_check_endpoint(app):
"""
建立健康檢查端點
這個端點用於監控系統和負載平衡器來檢查應用程式的健康狀態。
"""
@app.route('/health')
def health_check():
"""
健康檢查端點
檢查應用程式的各個元件是否正常運作。
"""
from app import db, cache
health_status = {
'status': 'healthy',
'checks': {}
}
# 檢查資料庫連線
try:
db.session.execute('SELECT 1')
health_status['checks']['database'] = 'ok'
except Exception as e:
health_status['status'] = 'unhealthy'
health_status['checks']['database'] = f'error: {str(e)}'
# 檢查快取連線
try:
cache.set('health_check', 'ok', timeout=1)
if cache.get('health_check') == 'ok':
health_status['checks']['cache'] = 'ok'
else:
raise Exception('Cache read/write failed')
except Exception as e:
health_status['status'] = 'unhealthy'
health_status['checks']['cache'] = f'error: {str(e)}'
# 根據健康狀態回傳適當的 HTTP 狀態碼
status_code = 200 if health_status['status'] == 'healthy' else 503
return health_status, status_code
監控系統應該收集多個層面的指標：應用程式層面的請求次數、回應時間、錯誤率；系統層面的 CPU 使用率、記憶體使用量、磁碟空間；業務層面的活躍使用者數、交易量等。健康檢查端點對於容器編排系統（如 Kubernetes）和負載平衡器來說特別重要，它們會定期呼叫這個端點來判斷應用程式是否可以接收流量。
測試策略
完善的測試是確保 API 品質的關鍵。測試應該涵蓋單元測試、整合測試和端對端測試，確保各個層面都經過驗證。
# tests/test_api.py
# 這個模組包含了 API 測試
import pytest
from app import create_app, db
from app.models import User, Item
from config import TestConfig
class TestConfig:
"""測試環境配置"""
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
JWT_SECRET_KEY = 'test-jwt-secret'
WTF_CSRF_ENABLED = False
@pytest.fixture
def app():
"""
建立測試用應用程式
這個 fixture 會在每個測試前建立一個全新的應用程式實例和資料庫。
"""
app = create_app(TestConfig)
with app.app_context():
db.create_all()
yield app
db.drop_all()
@pytest.fixture
def client(app):
"""建立測試客戶端"""
return app.test_client()
@pytest.fixture
def auth_headers(client, app):
"""
取得認證標頭
建立測試使用者並取得 JWT，用於需要認證的測試。
"""
with app.app_context():
# 建立測試使用者
from werkzeug.security import generate_password_hash
user = User(
username='testuser',
email='test@example.com',
password_hash=generate_password_hash('testpassword')
)
db.session.add(user)
db.session.commit()
# 登入取得 Token
response = client.post('/api/v1/auth/login', json={
'username': 'testuser',
'password': 'testpassword'
})
token = response.get_json()['access_token']
return {'Authorization': f'Bearer {token}'}
class TestAuthEndpoints:
"""認證端點測試"""
def test_register_success(self, client):
"""測試成功註冊"""
response = client.post('/api/v1/auth/register', json={
'username': 'newuser',
'email': 'new@example.com',
'password': 'securepassword123'
})
assert response.status_code == 201
assert 'user_id' in response.get_json()
def test_register_duplicate_username(self, client, app):
"""測試重複使用者名稱"""
# 先建立一個使用者
with app.app_context():
from werkzeug.security import generate_password_hash
user = User(
username='existinguser',
email='existing@example.com',
password_hash=generate_password_hash('password')
)
db.session.add(user)
db.session.commit()
# 嘗試用相同使用者名稱註冊
response = client.post('/api/v1/auth/register', json={
'username': 'existinguser',
'email': 'another@example.com',
'password': 'password123'
})
assert response.status_code == 409
assert '使用者名稱已被使用' in response.get_json()['message']
def test_login_success(self, client, app):
"""測試成功登入"""
# 建立測試使用者
with app.app_context():
from werkzeug.security import generate_password_hash
user = User(
username='logintest',
email='login@example.com',
password_hash=generate_password_hash('testpass')
)
db.session.add(user)
db.session.commit()
response = client.post('/api/v1/auth/login', json={
'username': 'logintest',
'password': 'testpass'
})
assert response.status_code == 200
data = response.get_json()
assert 'access_token' in data
assert 'refresh_token' in data
def test_login_invalid_credentials(self, client):
"""測試無效憑證"""
response = client.post('/api/v1/auth/login', json={
'username': 'nonexistent',
'password': 'wrongpassword'
})
assert response.status_code == 401
class TestItemEndpoints:
"""項目端點測試"""
def test_get_items_list(self, client):
"""測試取得項目列表"""
response = client.get('/api/v1/items')
assert response.status_code == 200
assert isinstance(response.get_json(), list)
def test_create_item_success(self, client, auth_headers):
"""測試成功新增項目"""
response = client.post('/api/v1/items',
json={
'name': 'Test Item',
'description': 'A test item',
'price': 99.99,
'quantity': 10
},
headers=auth_headers
)
assert response.status_code == 201
data = response.get_json()
assert data['name'] == 'Test Item'
assert data['price'] == 99.99
def test_create_item_unauthorized(self, client):
"""測試未授權新增項目"""
response = client.post('/api/v1/items', json={
'name': 'Test Item',
'price': 99.99,
'quantity': 10
})
assert response.status_code == 401
def test_create_item_invalid_data(self, client, auth_headers):
"""測試無效資料新增項目"""
# 缺少必填欄位
response = client.post('/api/v1/items',
json={
'description': 'Missing required fields'
},
headers=auth_headers
)
assert response.status_code == 400
def test_update_item_success(self, client, auth_headers, app):
"""測試成功更新項目"""
# 先建立一個項目
response = client.post('/api/v1/items',
json={
'name': 'Original Name',
'price': 50.0,
'quantity': 5
},
headers=auth_headers
)
item_id = response.get_json()['id']
# 更新項目
response = client.put(f'/api/v1/items/{item_id}',
json={
'name': 'Updated Name',
'price': 75.0
},
headers=auth_headers
)
assert response.status_code == 200
data = response.get_json()
assert data['name'] == 'Updated Name'
assert data['price'] == 75.0
def test_delete_item_success(self, client, auth_headers):
"""測試成功刪除項目"""
# 先建立一個項目
response = client.post('/api/v1/items',
json={
'name': 'To Delete',
'price': 10.0,
'quantity': 1
},
headers=auth_headers
)
item_id = response.get_json()['id']
# 刪除項目
response = client.delete(f'/api/v1/items/{item_id}',
headers=auth_headers
)
assert response.status_code == 200
# 確認已刪除
response = client.get(f'/api/v1/items/{item_id}')
assert response.status_code == 404
測試應該覆蓋正常流程和異常流程。正常流程測試確保功能按預期運作。異常流程測試確保應用程式能夠正確處理錯誤情況，如無效輸入、未授權存取、資源不存在等。使用 pytest 的 fixture 功能可以方便地設定測試前置條件，如建立測試資料庫、建立測試使用者等。
總結
本文深入探討了使用 Flask-RESTful 建立 REST API 的完整流程，從基礎的應用程式架構設計到進階的安全機制實作。Flask-RESTful 的資源類別設計使得程式碼組織更加清晰，reqparse 和 marshal_with 功能確保了輸入驗證和輸出格式化的一致性。
在安全方面，JWT 認證機制提供了無狀態、可擴展的認證解決方案，配合角色權限控管可以實現細緻的授權策略。速率限制、輸入驗證、HTML 消毒等措施則提供了額外的安全防護層。
效能最佳化涵蓋了多個層面：使用快取減少資料庫查詢、使用批次操作減少資料庫往返、使用高效的序列化庫加速資料處理、使用合適的 WSGI 伺服器配置充分利用硬體資源。
Docker 容器化簡化了部署流程，確保了不同環境間的一致性。多階段建置可以顯著減小映像檔大小，Docker Compose 提供了管理多容器應用程式的便利方式。
最後，監控和測試是確保生產系統健康運作的關鍵。指標收集和健康檢查端點提供了系統可觀測性，完善的測試覆蓋確保了程式碼品質。透過這些最佳實踐，開發者可以建立出安全、高效、可維護的企業級 Flask REST API 應用程式。
