上傳檔案到「/」
-- 指定專案資料夾 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
This commit is contained in:
250
employee_votes_api.py
Normal file
250
employee_votes_api.py
Normal file
@@ -0,0 +1,250 @@
|
||||
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)
|
||||
Reference in New Issue
Block a user