from flask import Flask, request, jsonify from flask_cors import CORS import mysql.connector from mysql.connector import Error app = Flask(__name__) CORS(app) # 資料庫連線資訊 db_config = { 'host': 'mysql.theaken.com', 'port': 33306, 'database': 'db_A021', 'user': 'A021', 'password': 'wJk1O5qtP7pf' } def get_db_connection(): try: connection = mysql.connector.connect(**db_config) return connection except Error as e: print(f"資料庫連線錯誤: {e}") return None @app.route('/v1/pizzas', methods=['GET']) def get_pizzas(): min_id = request.args.get('min_id', type=int) max_id = request.args.get('max_id', type=int) page = request.args.get('page', default=1, type=int) limit = request.args.get('limit', default=10, type=int) offset = (page - 1) * limit connection = get_db_connection() if not connection: return jsonify({"status": "error", "code": 500, "message": "資料庫連線失敗"}), 500 try: cursor = connection.cursor(dictionary=True) # 基礎查詢 query = "SELECT * FROM pizzas WHERE 1=1" params = [] # 添加篩選條件 if min_id is not None: query += " AND id >= %s" params.append(min_id) if max_id is not None: query += " AND id <= %s" params.append(max_id) # 計算總數 count_query = "SELECT COUNT(*) as total FROM (" + query + ") as t" cursor.execute(count_query, params) total = cursor.fetchone()['total'] # 添加分頁 query += " LIMIT %s OFFSET %s" params.extend([limit, offset]) cursor.execute(query, params) pizzas = cursor.fetchall() meta = { "total": total, "page": page, "limit": limit, "has_next": (page * limit) < total } return jsonify({ "status": "success", "code": 200, "message": "", "data": pizzas, "meta": meta }) except Error as e: return jsonify({"status": "error", "code": 500, "message": str(e)}), 500 finally: if connection.is_connected(): cursor.close() connection.close() @app.route('/v1/pizzas/', methods=['GET']) def get_pizza(pizza_id): connection = get_db_connection() if not connection: return jsonify({"status": "error", "code": 500, "message": "資料庫連線失敗"}), 500 try: cursor = connection.cursor(dictionary=True) cursor.execute("SELECT * FROM pizzas WHERE id = %s", (pizza_id,)) pizza = cursor.fetchone() if not pizza: return jsonify({"status": "error", "code": 404, "message": "披薩不存在"}), 404 return jsonify({"status": "success", "code": 200, "message": "", "data": pizza}) except Error as e: return jsonify({"status": "error", "code": 500, "message": str(e)}), 500 finally: if connection.is_connected(): cursor.close() connection.close() @app.route('/v1/pizzas', methods=['POST']) def create_pizza(): data = request.get_json() # 驗證必要欄位 if not all(key in data for key in ['name', 'size', 'price']): return jsonify({"status": "error", "code": 400, "message": "缺少必要欄位: name, size, price"}), 400 connection = get_db_connection() if not connection: return jsonify({"status": "error", "code": 500, "message": "資料庫連線失敗"}), 500 try: cursor = connection.cursor(dictionary=True) query = "INSERT INTO pizzas (name, size, price) VALUES (%s, %s, %s)" cursor.execute(query, (data['name'], data['size'], data['price'])) connection.commit() # 獲取新建立的披薩 pizza_id = cursor.lastrowid cursor.execute("SELECT * FROM pizzas WHERE id = %s", (pizza_id,)) new_pizza = cursor.fetchone() return jsonify({ "status": "success", "code": 201, "message": "披薩建立成功", "data": new_pizza }), 201 except Error as e: return jsonify({"status": "error", "code": 500, "message": str(e)}), 500 finally: if connection.is_connected(): cursor.close() connection.close() @app.route('/v1/pizzas/', methods=['PATCH']) def update_pizza(pizza_id): # 從URL參數獲取要更新的欄位 update_name = request.args.get('name') update_size = request.args.get('size') update_price = request.args.get('price', type=float) connection = get_db_connection() if not connection: return jsonify({"status": "error", "code": 500, "message": "資料庫連線失敗"}), 500 try: cursor = connection.cursor(dictionary=True) # 檢查披薩是否存在 cursor.execute("SELECT * FROM pizzas WHERE id = %s", (pizza_id,)) pizza = cursor.fetchone() if not pizza: return jsonify({"status": "error", "code": 404, "message": "披薩不存在"}), 404 # 構建更新語句 set_clauses = [] params = [] if update_name is not None: set_clauses.append("name = %s") params.append(update_name) if update_size is not None: set_clauses.append("size = %s") params.append(update_size) if update_price is not None: set_clauses.append("price = %s") params.append(update_price) # 如果沒有提供任何可更新參數 if not set_clauses: return jsonify({"status": "error", "code": 400, "message": "至少需要提供一個更新參數: name, size, price"}), 400 query = "UPDATE pizzas SET " + ", ".join(set_clauses) + " WHERE id = %s" params.append(pizza_id) cursor.execute(query, params) connection.commit() # 獲取更新後的披薩 cursor.execute("SELECT * FROM pizzas WHERE id = %s", (pizza_id,)) updated_pizza = cursor.fetchone() return jsonify({ "status": "success", "code": 200, "message": "披薩更新成功", "data": updated_pizza }) except Error as e: return jsonify({"status": "error", "code": 500, "message": str(e)}), 500 finally: if connection.is_connected(): cursor.close() connection.close() @app.route('/v1/pizzas/', methods=['DELETE']) def delete_pizza(pizza_id): connection = get_db_connection() if not connection: return jsonify({"status": "error", "code": 500, "message": "資料庫連線失敗"}), 500 try: cursor = connection.cursor() # 檢查披薩是否存在 cursor.execute("SELECT id FROM pizzas WHERE id = %s", (pizza_id,)) if not cursor.fetchone(): return jsonify({"status": "error", "code": 404, "message": "披薩不存在"}), 404 cursor.execute("DELETE FROM pizzas WHERE id = %s", (pizza_id,)) connection.commit() return '', 204 except Error as e: return jsonify({"status": "error", "code": 500, "message": str(e)}), 500 finally: if connection.is_connected(): cursor.close() connection.close() if __name__ == '__main__': app.run(debug=True)