91771
314474a682
-- 指定專案資料夾 cd /employee_votes -- gitea 初始化 git init -- 建立註解 git add . git commit -m "Initial commit" -- 切換到 main 分支 git branch -M main -- 上傳檔案到 gitea git remote add origin https://github.com/91771/<REPO>.git git push -u origin main
250 lines
8.4 KiB
Python
250 lines
8.4 KiB
Python
from flask import Flask, request, jsonify
|
|
from flask_cors import CORS
|
|
import mysql.connector
|
|
|
|
app = Flask(__name__)
|
|
CORS(app, resources={r"/v1/*": {"origins": "http://localhost:3000"}})
|
|
|
|
# Database connection details from DB_connection.txt
|
|
DB_CONFIG = {
|
|
'host': 'mysql.theaken.com',
|
|
'port': 33306,
|
|
'database': 'db_A019',
|
|
'user': 'A019',
|
|
'password': '9wvKEkxBzVca'
|
|
}
|
|
|
|
def get_db_connection():
|
|
try:
|
|
conn = mysql.connector.connect(**DB_CONFIG)
|
|
return conn
|
|
except mysql.connector.Error as err:
|
|
print(f"Error connecting to database: {err}")
|
|
return None
|
|
|
|
def format_response(status, code, message, data=None):
|
|
return jsonify({
|
|
"status": status,
|
|
"code": code,
|
|
"message": message,
|
|
"data": data
|
|
}), code
|
|
|
|
def format_error(code, message):
|
|
return jsonify({
|
|
"status": "error",
|
|
"code": code,
|
|
"message": message
|
|
}), code
|
|
|
|
@app.route('/v1/employee_votes', methods=['GET'])
|
|
def get_employee_votes():
|
|
conn = get_db_connection()
|
|
if conn is None:
|
|
return format_error(500, "Database connection failed")
|
|
cursor = conn.cursor(dictionary=True)
|
|
|
|
query = "SELECT * FROM employee_votes WHERE 1=1"
|
|
params = []
|
|
|
|
if 'menu_item_id' in request.args:
|
|
query += " AND menu_item_id = %s"
|
|
params.append(request.args['menu_item_id'])
|
|
if 'emp_id' in request.args:
|
|
query += " AND emp_id = %s"
|
|
params.append(request.args['emp_id'])
|
|
if 'emp_name' in request.args:
|
|
query += " AND emp_name = %s"
|
|
params.append(request.args['emp_name'])
|
|
if 'Order_Date' in request.args:
|
|
query += " AND Order_Date = %s"
|
|
params.append(request.args['Order_Date'])
|
|
if 'is_active' in request.args:
|
|
query += " AND is_active = %s"
|
|
params.append(request.args['is_active'])
|
|
|
|
try:
|
|
cursor.execute(query, tuple(params))
|
|
employee_votes = cursor.fetchall()
|
|
meta = {"count": len(employee_votes)}
|
|
return format_response("success", 200, "Employee votes retrieved successfully", {"employee_votes": employee_votes, "meta": meta})
|
|
except mysql.connector.Error as err:
|
|
return format_error(500, f"Error retrieving employee votes: {err}")
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
@app.route('/v1/employee_votes/<int:id>', methods=['GET'])
|
|
def get_employee_vote_by_id(id):
|
|
conn = get_db_connection()
|
|
if conn is None:
|
|
return format_error(500, "Database connection failed")
|
|
cursor = conn.cursor(dictionary=True)
|
|
|
|
query = "SELECT * FROM employee_votes WHERE id = %s"
|
|
try:
|
|
cursor.execute(query, (id,))
|
|
employee_vote = cursor.fetchone()
|
|
if employee_vote:
|
|
return format_response("success", 200, "Employee vote retrieved successfully", employee_vote)
|
|
else:
|
|
return format_error(404, "Employee vote not found")
|
|
except mysql.connector.Error as err:
|
|
return format_error(500, f"Error retrieving employee vote: {err}")
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
@app.route('/v1/employee_votes', methods=['POST'])
|
|
def create_employee_vote():
|
|
conn = get_db_connection()
|
|
if conn is None:
|
|
return format_error(500, "Database connection failed")
|
|
cursor = conn.cursor(dictionary=True)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return format_error(400, "Invalid JSON data")
|
|
|
|
required_fields = ['menu_item_id', 'emp_id', 'emp_name', 'Order_Date']
|
|
if not all(field in data for field in required_fields):
|
|
return format_error(400, "Missing required fields")
|
|
|
|
menu_item_id = data['menu_item_id']
|
|
emp_id = data['emp_id']
|
|
emp_name = data['emp_name']
|
|
order_date = data['Order_Date']
|
|
|
|
insert_query = "INSERT INTO employee_votes (menu_item_id, emp_id, emp_name, Order_Date) VALUES (%s, %s, %s, %s)"
|
|
try:
|
|
cursor.execute(insert_query, (menu_item_id, emp_id, emp_name, order_date))
|
|
conn.commit()
|
|
new_id = cursor.lastrowid
|
|
new_vote = {"id": new_id, "menu_item_id": menu_item_id, "emp_id": emp_id, "emp_name": emp_name, "Order_Date": order_date, "is_active": 1}
|
|
return format_response("success", 201, "Employee vote created successfully", new_vote)
|
|
except mysql.connector.Error as err:
|
|
conn.rollback()
|
|
return format_error(500, f"Error creating employee vote: {err}")
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
@app.route('/v1/employee_votes/<int:id>', methods=['PATCH'])
|
|
def update_employee_vote(id):
|
|
conn = get_db_connection()
|
|
if conn is None:
|
|
return format_error(500, "Database connection failed")
|
|
cursor = conn.cursor(dictionary=True)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return format_error(400, "Invalid JSON data")
|
|
|
|
update_fields = []
|
|
params = []
|
|
|
|
if 'menu_item_id' in data:
|
|
update_fields.append("menu_item_id = %s")
|
|
params.append(data['menu_item_id'])
|
|
if 'is_active' in data:
|
|
update_fields.append("is_active = %s")
|
|
params.append(data['is_active'])
|
|
|
|
if not update_fields:
|
|
return format_error(400, "No fields to update")
|
|
|
|
update_query = "UPDATE employee_votes SET " + ", ".join(update_fields) + " WHERE id = %s"
|
|
params.append(id)
|
|
|
|
try:
|
|
cursor.execute(update_query, tuple(params))
|
|
conn.commit()
|
|
if cursor.rowcount == 0:
|
|
return format_error(404, "Employee vote not found")
|
|
|
|
# Retrieve updated data
|
|
cursor.execute("SELECT * FROM employee_votes WHERE id = %s", (id,))
|
|
updated_vote = cursor.fetchone()
|
|
return format_response("success", 200, "Employee vote updated successfully", updated_vote)
|
|
except mysql.connector.Error as err:
|
|
conn.rollback()
|
|
return format_error(500, f"Error updating employee vote: {err}")
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
@app.route('/v1/employee_votes/by_emp_date', methods=['PATCH'])
|
|
def update_employee_vote_by_emp_date():
|
|
conn = get_db_connection()
|
|
if conn is None:
|
|
return format_error(500, "Database connection failed")
|
|
cursor = conn.cursor(dictionary=True)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return format_error(400, "Invalid JSON data")
|
|
|
|
required_fields = ['emp_id', 'Order_Date']
|
|
if not all(field in data for field in required_fields):
|
|
return format_error(400, "Missing required fields: emp_id and Order_Date")
|
|
|
|
emp_id = data['emp_id']
|
|
order_date = data['Order_Date']
|
|
|
|
update_fields = []
|
|
params = []
|
|
|
|
if 'menu_item_id' in data:
|
|
update_fields.append("menu_item_id = %s")
|
|
params.append(data['menu_item_id'])
|
|
if 'is_active' in data:
|
|
update_fields.append("is_active = %s")
|
|
params.append(data['is_active'])
|
|
|
|
if not update_fields:
|
|
return format_error(400, "No fields to update")
|
|
|
|
update_query = "UPDATE employee_votes SET " + ", ".join(update_fields) + " WHERE emp_id = %s AND Order_Date = %s"
|
|
params.append(emp_id)
|
|
params.append(order_date)
|
|
|
|
try:
|
|
cursor.execute(update_query, tuple(params))
|
|
conn.commit()
|
|
if cursor.rowcount == 0:
|
|
return format_error(404, "Employee vote not found for the given emp_id and Order_Date")
|
|
|
|
# Retrieve updated data
|
|
cursor.execute("SELECT * FROM employee_votes WHERE emp_id = %s AND Order_Date = %s", (emp_id, order_date))
|
|
updated_votes = cursor.fetchall()
|
|
return format_response("success", 200, "Employee vote updated successfully", updated_votes)
|
|
except mysql.connector.Error as err:
|
|
conn.rollback()
|
|
return format_error(500, f"Error updating employee vote: {err}")
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
@app.route('/v1/employee_votes/<int:id>', methods=['DELETE'])
|
|
def delete_employee_vote(id):
|
|
conn = get_db_connection()
|
|
if conn is None:
|
|
return format_error(500, "Database connection failed")
|
|
cursor = conn.cursor(dictionary=True)
|
|
|
|
delete_query = "DELETE FROM employee_votes WHERE id = %s"
|
|
try:
|
|
cursor.execute(delete_query, (id,))
|
|
conn.commit()
|
|
if cursor.rowcount == 0:
|
|
return format_error(404, "Employee vote not found")
|
|
return format_response("success", 204, "Employee vote deleted successfully")
|
|
except mysql.connector.Error as err:
|
|
conn.rollback()
|
|
return format_error(500, f"Error deleting employee vote: {err}")
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
if __name__ == '__main__':
|
|
app.run(debug=True, port=5000) |