231 lines
7.5 KiB
Python
231 lines
7.5 KiB
Python
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/<int:pizza_id>', 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/<int:pizza_id>', 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/<int:pizza_id>', 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) |