這個 Flask 電商 Web API 從建立應用程式框架開始，逐步加入產品目錄、購物車和結帳功能。示範程式碼使用 SQLite 資料函式庫，包含資料函式庫連線、表格建立和資料操作。為了提升程式碼品質，後續章節著重程式碼重構、模組化設計和安全性強化，例如將資料函式庫操作獨立成模組，並使用引數化查詢防止 SQL 資料隱碼攻擊。最終版本程式碼更簡潔易懂，也更安全可靠，適合實際專案應用。
使用 Flask 建立電子商務網站的 Web API
本章節將指導如何使用 Flask 建立一個電子商務網站的 Web API。該 API 將包括產品目錄、購物車和結帳流程等功能。
步驟 1：建立 Web API 的基本結構
首先，我們需要建立一個基本的 Flask 應用程式。以下是一個簡單的範例程式碼：
from flask import Flask, jsonify
app = Flask(__name__)
@app.route('/')
def index():
return jsonify({'message': 'Hello World!'})
@app.route('/products')
def products():
products = [
{'id': 1, 'name': 'Product 1', 'price': 10.99},
{'id': 2, 'name': 'Product 2', 'price': 5.99},
{'id': 3, 'name': 'Product 3', 'price': 7.99}
]
return jsonify(products)
@app.route('/cart')
def cart():
return jsonify({'message': 'Cart'})
@app.route('/checkout')
def checkout():
return jsonify({'message': 'Checkout'})
if __name__ == '__main__':
app.run(debug=True, port=5000)
程式碼解析：
- 使用
Flask建立一個 Web 應用程式。 - 定義了四個路由：
/、/products、/cart和/checkout。 - 使用
jsonify將資料轉換為 JSON 格式傳回給客戶端。
步驟 2：連線資料函式庫並讀寫資料
接下來，我們需要將資料儲存在資料函式庫中。以下是一個使用 SQLite 資料函式庫的範例：
import sqlite3
from flask import Flask, jsonify
app = Flask(__name__)
# 連線資料函式庫
conn = sqlite3.connect('example.db')
# 建立 products 表格
conn.execute('''CREATE TABLE IF NOT EXISTS products
(id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
price REAL NOT NULL);''')
# 插入初始資料
conn.execute("INSERT INTO products (name, price) VALUES (?, ?)", ('Product 1', 10.99))
conn.execute("INSERT INTO products (name, price) VALUES (?, ?)", ('Product 2', 5.99))
conn.execute("INSERT INTO products (name, price) VALUES (?, ?)", ('Product 3', 7.99))
conn.commit()
@app.route('/products')
def products():
cursor = conn.execute("SELECT id, name, price FROM products")
products = []
for row in cursor:
product = {'id': row[0], 'name': row[1], 'price': row[2]}
products.append(product)
return jsonify(products)
if __name__ == '__main__':
app.run(debug=True, port=5000)
程式碼解析：
- 使用
sqlite3連線 SQLite 資料函式庫。 - 建立了一個名為
products的表格，並插入了一些初始資料。 - 在
/products路由中，從資料函式庫讀取產品資料並傳回 JSON 格式的資料。
改進方向
- 新增測試：為 Web API 編寫單元測試和整合測試，以確保其正確性和穩定性。
- 檔案化：為 Web API 編寫檔案，以便其他開發者瞭解如何使用它。
利用Web API構建後端系統
提升資料函式庫互動功能
在前面的範例中，我們成功建立了一個基本的Flask應用程式，並實作了與SQLite資料函式庫的連線。然而，程式碼仍存在一些需要改進的地方，例如：只有/products路由能夠正確讀取資料函式庫中的資料，而其他路由尚未實作資料函式庫的讀寫功能。
程式碼最佳化與資料函式庫操作
為了提升程式碼的功能性，我們需要對現有的程式碼進行最佳化，確保所有路由都能正確地與資料函式庫進行互動。以下是一個改進後的版本：
import sqlite3
import json
from flask import Flask, jsonify, request
# 建立Flask應用程式
app = Flask(__name__)
# 連線至SQLite資料函式庫
def get_db_connection():
conn = sqlite3.connect('example.db')
conn.row_factory = sqlite3.Row
return conn
# 建立所需的資料表
def create_tables():
conn = get_db_connection()
conn.execute('''CREATE TABLE IF NOT EXISTS products
(id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
price REAL NOT NULL);''')
conn.execute('''CREATE TABLE IF NOT EXISTS cart
(id INTEGER PRIMARY KEY AUTOINCREMENT,
product_id INTEGER NOT NULL,
quantity INTEGER NOT NULL);''')
conn.execute('''CREATE TABLE IF NOT EXISTS checkout
(id INTEGER PRIMARY KEY AUTOINCREMENT,
cart_id INTEGER NOT NULL,
total REAL NOT NULL);''')
conn.commit()
conn.close()
create_tables()
# 首頁路由
@app.route('/')
def index():
return jsonify({'message': 'Hello World!'})
#### 內容解密：
1. **首頁路由**：此路由傳回一個簡單的JSON訊息，表明應用程式執行正常。
2. **資料函式庫連線函式**：`get_db_connection`函式負責建立與SQLite資料函式庫的連線，並設定`row_factory`為`sqlite3.Row`，以便於以字典形式存取查詢結果。
# 產品目錄路由
@app.route('/products')
def products():
conn = get_db_connection()
products = conn.execute("SELECT id, name, price FROM products").fetchall()
conn.close()
return jsonify([dict(row) for row in products])
#### 內容解密：
1. **產品目錄路由**：此路由查詢`products`表中的所有產品，並將結果以JSON格式傳回。
2. **查詢結果處理**：使用列表推導式將查詢結果轉換為字典列表，以便於轉換為JSON格式。
# 購物車路由
@app.route('/cart')
def cart():
conn = get_db_connection()
cart_items = conn.execute("SELECT id, product_id, quantity FROM cart").fetchall()
conn.close()
return jsonify([dict(row) for row in cart_items])
#### 內容解密：
1. **購物車路由**：此路由查詢`cart`表中的所有購物車專案，並將結果以JSON格式傳回。
2. **查詢結果處理**：與產品目錄路由類別似，將查詢結果轉換為字典列表後傳回。
# 新增產品至購物車路由
@app.route('/cart', methods=['POST'])
def add_to_cart():
conn = get_db_connection()
data = request.json
product_id = data['productId']
quantity = data['quantity']
# 檢查購物車是否已存在
existing_cart_item = conn.execute("SELECT id FROM cart WHERE product_id = ?", (product_id,)).fetchone()
if existing_cart_item:
# 更新現有的購物車專案
conn.execute("UPDATE cart SET quantity = quantity + ? WHERE product_id = ?", (quantity, product_id))
else:
# 新增新的購物車專案
conn.execute("INSERT INTO cart (product_id, quantity) VALUES (?, ?)", (product_id, quantity))
conn.commit()
conn.close()
return jsonify({'message': 'Added to cart'})
#### 內容解密：
1. **新增產品至購物車**：此路由接收POST請求，將產品新增到購物車或更新現有產品的數量。
2. **檢查購物車專案**：查詢`cart`表中是否已存在相同的產品，若存在則更新數量，否則新增一筆記錄。
# 結帳路由
@app.route('/checkout', methods=['POST'])
def checkout():
conn = get_db_connection()
data = request.json
cart_id = data['cartId']
# 假設總價計算邏輯在此處進行，實際應用中應根據業務邏輯進行調整
total = 0.0 # 示例總價
conn.execute("INSERT INTO checkout (cart_id, total) VALUES (?, ?)", (cart_id, total))
conn.commit()
conn.close()
return jsonify({'message': 'Checkout'})
#### 內容解密：
1. **結帳路由**：此路由接收POST請求，完成結帳操作，將購物車ID和總價記錄到`checkout`表中。
2. **總價計算**：在實際應用中，應根據業務邏輯計算總價，此處僅為示例。
if __name__ == '__main__':
app.run(debug=True, port=5000)
程式碼最佳化與提升
透過上述最佳化，我們實作了以下改進：
- 完整資料函式庫操作：所有路由均能正確地與資料函式庫進行互動，包括讀取和寫入操作。
- 最佳化資料函式庫連線管理：使用
get_db_connection函式統一管理資料函式庫連線，確保連線的正確建立和關閉。 - 提升程式碼可讀性：透過清晰的註解和結構化的程式碼，提升了可讀性和可維護性。
未來改進方向
- 安全性增強：增加輸入驗證和錯誤處理機制，提升應用程式的安全性和穩定性。
- 業務邏輯完善：根據實際業務需求，完善結帳邏輯和購物車管理功能。
- 效能最佳化：考慮使用連線池等技術，最佳化資料函式庫操作的效能。
建構具備 Web API 的後端系統
提升程式碼品質與安全性
在開發後端系統時，我們需要確保程式碼不僅能正常運作，還需具備可維護性、安全性和擴充性。以下將逐步分析與改程式式碼。
原始程式碼分析
原始程式碼存在多個問題，包括：
- 程式碼重複性高，未遵循 DRY（Don’t Repeat Yourself）原則。
- 缺乏驗證與授權機制。
- 無檔案說明。
- 資安措施不足，容易受到 SQL 資料隱碼攻擊（SQL Injection）。
程式碼改善步驟
- 執行程式碼並測試：首先執行現有程式碼，觀察錯誤並進行初步修正。
- 架構與設計最佳化：將程式碼分割至不同檔案，提升可維護性。
- 加入驗證、授權及安全措施：確保系統安全性，防止潛在攻擊。
改善後的程式碼架構
將資料函式庫相關操作獨立至 db.py 檔案，並在 app.py 中參照。以下為改善後的程式碼範例：
db.py
import sqlite3
def get_db_connection():
conn = sqlite3.connect('example.db')
conn.row_factory = sqlite3.Row
return conn
def get_products():
conn = get_db_connection()
products = conn.execute("SELECT id, name, price FROM products").fetchall()
conn.close()
return [dict(row) for row in products]
def get_cart():
conn = get_db_connection()
cart = conn.execute("SELECT id, product_id, quantity FROM cart").fetchall()
conn.close()
return [dict(row) for row in cart]
def add_to_cart(product_id, quantity):
conn = get_db_connection()
cursor = conn.execute("SELECT id FROM cart WHERE product_id = ?", (product_id,))
cart_item = cursor.fetchone()
if cart_item:
conn.execute("UPDATE cart SET quantity = quantity + ? WHERE product_id = ?", (quantity, product_id))
else:
conn.execute("INSERT INTO cart (product_id, quantity) VALUES (?, ?)", (product_id, quantity))
conn.commit()
conn.close()
def checkout(cart_id):
conn = get_db_connection()
conn.execute("INSERT INTO checkout (cart_id, total) VALUES (?, ?)", (cart_id, 0))
conn.commit()
conn.close()
app.py
import flask
import db
app = flask.Flask(__name__)
@app.route('/')
def index():
return flask.jsonify({'message': 'Hello World!'})
@app.route('/products')
def products():
products = db.get_products()
return flask.jsonify(products)
@app.route('/cart')
def cart():
cart = db.get_cart()
return flask.jsonify(cart)
@app.route('/cart', methods=['POST'])
def add_to_cart():
product_id = flask.request.json['productId']
quantity = flask.request.json['quantity']
db.add_to_cart(product_id, quantity)
return flask.jsonify({'message': 'Added to cart'})
@app.route('/checkout', methods=['POST'])
def checkout():
cart_id = flask.request.json['cartId']
db.checkout(cart_id)
return flask.jsonify({'message': 'Checkout successful'})
if __name__ == '__main__':
app.run(debug=True, port=5000)
#### 內容解密：
- 將資料函式庫連線操作封裝於
get_db_connection函式，確保每次操作後關閉連線，避免資源浪費。 - 資料查詢結果透過
fetchall()取得，並轉換為字典格式，便於 JSON 回傳。 add_to_cart函式檢查購物車中是否已有相同商品，若有則更新數量，否則新增專案。checkout函式將購物車 ID 寫入結帳記錄，並初始化總金額為 0。
重構與安全性提升
- 將資料函式庫操作獨立至
db.py，提升程式碼模組化與可維護性。 - 使用引數化查詢，防止 SQL 資料隱碼攻擊。
- 確保每次資料函式庫操作後關閉連線，釋放資源。