C0908/flask_pizza_api.py

from flask import Flask, request, jsonify
from flask_cors import CORS
import mysql.connector
from mysql.connector import errorcode

app = Flask(__name__)
CORS(app)

# 資料庫連線資訊
DB_HOST = "mysql.theaken.com"
DB_PORT = 33306
DB_NAME = "db_A018"
DB_USER = "A018"
DB_PASSWORD = "4MQYkJRYtyLE"

# 統一回應格式
def success_response(data=None, code=200, message="Success"):
    return jsonify({"status": "success", "code": code, "message": message, "data": data}), code

def error_response(code, message):
    return jsonify({"status": "error", "code": code, "message": message}), code

# 建立資料庫連線
def get_db_connection():
    return mysql.connector.connect(
        host=DB_HOST,
        port=DB_PORT,
        user=DB_USER,
        password=DB_PASSWORD,
        database=DB_NAME
    )

@app.route("/v1/pizza", methods=["GET"])
def get_pizzas():
    try:
        min_price = request.args.get("min_price")
        max_price = request.args.get("max_price")
        page = int(request.args.get("page", 1))
        limit = int(request.args.get("limit", 10))
        offset = (page - 1) * limit

        conn = get_db_connection()
        cursor = conn.cursor(dictionary=True)

        query = "SELECT * FROM pizza WHERE 1=1"
        params = []

        if min_price:
            query += " AND price >= %s"
            params.append(min_price)
        if max_price:
            query += " AND price <= %s"
            params.append(max_price)

        query += " LIMIT %s OFFSET %s"
        params.extend([limit, offset])

        cursor.execute(query, params)
        pizzas = cursor.fetchall()

        cursor.execute("SELECT COUNT(*) as total FROM pizza")
        total = cursor.fetchone()["total"]

        meta = {"page": page, "limit": limit, "total": total}
        return success_response({"pizzas": pizzas, "meta": meta})

    except Exception as e:
        return error_response(500, str(e))
    finally:
        if 'cursor' in locals():
            cursor.close()
        if 'conn' in locals():
            conn.close()

@app.route("/v1/pizza/<int:id>", methods=["GET"])
def get_pizza(id):
    try:
        conn = get_db_connection()
        cursor = conn.cursor(dictionary=True)

        cursor.execute("SELECT * FROM pizza WHERE id = %s", (id,))
        pizza = cursor.fetchone()

        if not pizza:
            return error_response(404, "Pizza not found")

        return success_response(pizza)

    except Exception as e:
        return error_response(500, str(e))
    finally:
        if 'cursor' in locals():
            cursor.close()
        if 'conn' in locals():
            conn.close()

@app.route("/v1/pizza", methods=["POST"])
def create_pizza():
    try:
        data = request.get_json()
        name = data.get("name")
        size = data.get("size")
        price = data.get("price")

        if not all([name, size, price]):
            return error_response(400, "Missing required fields")

        conn = get_db_connection()
        cursor = conn.cursor(dictionary=True)

        cursor.execute(
            "INSERT INTO pizza (name, size, price) VALUES (%s, %s, %s)",
            (name, size, price)
        )
        conn.commit()

        new_id = cursor.lastrowid
        cursor.execute("SELECT * FROM pizza WHERE id = %s", (new_id,))
        new_pizza = cursor.fetchone()

        return success_response(new_pizza, 201, "Pizza created successfully")

    except Exception as e:
        return error_response(500, str(e))
    finally:
        if 'cursor' in locals():
            cursor.close()
        if 'conn' in locals():
            conn.close()

@app.route("/v1/pizza/<int:id>", methods=["PATCH"])
def update_pizza(id):
    try:
        data = request.get_json()
        fields = []
        params = []

        for key in ["name", "size", "price"]:
            if key in data:
                fields.append(f"{key} = %s")
                params.append(data[key])

        if not fields:
            return error_response(400, "No fields to update")

        params.append(id)

        conn = get_db_connection()
        cursor = conn.cursor(dictionary=True)

        cursor.execute(f"UPDATE pizza SET {', '.join(fields)} WHERE id = %s", params)
        conn.commit()

        cursor.execute("SELECT * FROM pizza WHERE id = %s", (id,))
        updated_pizza = cursor.fetchone()

        if not updated_pizza:
            return error_response(404, "Pizza not found")

        return success_response(updated_pizza, 200, "Pizza updated successfully")

    except Exception as e:
        return error_response(500, str(e))
    finally:
        if 'cursor' in locals():
            cursor.close()
        if 'conn' in locals():
            conn.close()

@app.route("/v1/pizza/<int:id>", methods=["DELETE"])
def delete_pizza(id):
    try:
        conn = get_db_connection()
        cursor = conn.cursor()

        cursor.execute("DELETE FROM pizza WHERE id = %s", (id,))
        conn.commit()

        if cursor.rowcount == 0:
            return error_response(404, "Pizza not found")

        return success_response(None, 204, "Pizza deleted successfully")

    except Exception as e:
        return error_response(500, str(e))
    finally:
        if 'cursor' in locals():
            cursor.close()
        if 'conn' in locals():
            conn.close()

if __name__ == "__main__":
    app.run(debug=True)