Flask - 初探數據存儲
目標
- 使用 Python 字典模擬數據庫
- 實現完整的 CRUD 操作
- 保持 API 的結構化響應和錯誤處理
步驟
- 準備環境
- 繼續使用第5天的模塊化結構,確保您在
flask_api/
目錄並激活虛擬環境:1 2
# Windows: flask_api_env\Scripts\activate # macOS/Linux: source flask_api_env/bin/activate
- 繼續使用第5天的模塊化結構,確保您在
- 設計數據存儲
- 我們將使用一個 Python 字典來模擬數據庫,存儲 “items” 的數據。字典的鍵是項目 ID,值是項目詳情。
- 更新應用
- 修改 app/routes/items.py,實現完整的 CRUD 操作:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67
from flask import Blueprint, jsonify, request, abort items_bp = Blueprint('items', __name__) # 模擬數據庫(使用字典) items_db = { 1: {'id': 1, 'name': 'Item 1', 'price': 10.0}, 2: {'id': 2, 'name': 'Item 2', 'price': 20.0} } # GET - 獲取所有項目 @items_bp.route('/items', methods=['GET']) def get_items(): max_price = request.args.get('max_price', type=float) filtered_items = list(items_db.values()) if max_price is not None: filtered_items = [item for item in filtered_items if item['price'] <= max_price] return jsonify({'items': filtered_items}) # GET - 獲取單個項目 @items_bp.route('/items/<int:item_id>', methods=['GET']) def get_item(item_id): item = items_db.get(item_id) if not item: abort(404, description='Item not found') return jsonify(item) # POST - 創建新項目 @items_bp.route('/items', methods=['POST']) def create_item(): if not request.is_json: abort(400, description='Request must be JSON') data = request.get_json() if 'name' not in data or 'price' not in data: abort(400, description='Missing name or price') if not isinstance(data['price'], (int, float)) or data['price'] < 0: abort(400, description='Price must be a non-negative number') new_id = max(items_db.keys(), default=0) + 1 new_item = { 'id': new_id, 'name': data['name'], 'price': data['price'] } items_db[new_id] = new_item return jsonify(new_item), 201 # PUT - 更新項目 @items_bp.route('/items/<int:item_id>', methods=['PUT']) def update_item(item_id): if item_id not in items_db: abort(404, description='Item not found') if not request.is_json: abort(400, description='Request must be JSON') data = request.get_json() if 'price' in data and (not isinstance(data['price'], (int, float)) or data['price'] < 0): abort(400, description='Price must be a non-negative number') item = items_db[item_id] item.update({k: v for k, v in data.items() if k in ['name', 'price']}) return jsonify(item), 200 # DELETE - 刪除項目 @items_bp.route('/items/<int:item_id>', methods=['DELETE']) def delete_item(item_id): if item_id not in items_db: abort(404, description='Item not found') del items_db[item_id] return jsonify({'message': 'Item deleted'}), 200
- 代碼解釋:
items_db
:使用字典替代列表,鍵為 ID,方便查找和刪除。max(items_db.keys(), default=0) + 1
:生成新的唯一 ID。- 驗證
price
:確保價格是非負數。
- 修改 app/routes/items.py,實現完整的 CRUD 操作:
- 確保錯誤處理有效
- app/__init__.py 中的錯誤處理器(第5天代碼)應保持不變,會自動處理
abort
拋出的錯誤。
- app/__init__.py 中的錯誤處理器(第5天代碼)應保持不變,會自動處理
- 運行應用
- 運行:
1
python run.py
- 運行:
- 測試 CRUD 操作
- 使用 Postman 測試以下端點:
- GET /api/v1/items:
- 預期響應:
{"items": [{"id": 1, "name": "Item 1", "price": 10.0}, {"id": 2, "name": "Item 2", "price": 20.0}]}
- 預期響應:
- GET /api/v1/items/1:
- 預期響應:
{"id": 1, "name": "Item 1", "price": 10.0}
- 預期響應:
- POST /api/v1/items:
- Body:
{"name": "Item 3", "price": 30.0}
- 預期響應:
{"id": 3, "name": "Item 3", "price": 30.0}
- Body:
- PUT /api/v1/items/1:
- Body:
{"price": 15.0}
- 預期響應:
{"id": 1, "name": "Item 1", "price": 15.0}
- Body:
- DELETE /api/v1/items/2:
- 預期響應:
{"message": "Item deleted"}
- 預期響應:
- 錯誤測試:
- POST 負價格:
{"name": "Item 4", "price": -5}
,應返回 400。
- POST 負價格:
- GET /api/v1/items:
- 使用 Postman 測試以下端點:
- 作業
- 添加一個查詢參數
name
到 GET/items
,支持按名稱過濾項目。 - 在
items_db
中添加一個新字段(如category
),並更新 CRUD 操作以支持它。
- 添加一個查詢參數
注意事項
- 這是記憶體數據,重啟應用後數據會丟失,下一週我們將引入持久化數據庫。
- 確保輸入驗證嚴謹,避免無效數據進入
items_db
。
本文章以 CC BY 4.0 授權