-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathstatusList.py
More file actions
79 lines (65 loc) · 2.84 KB
/
statusList.py
File metadata and controls
79 lines (65 loc) · 2.84 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
# -*- coding: utf-8 -*-
"""
Created on Thu Feb 1 12:57:07 2024
@author: Yusuf Sanni
This is a Python API for getting the list of statuses available for use from the database.
In other words, it checks the statuses of the different available infrastructure to know which
is vacant or occupied. It pulls this information from a table called status_lookup in the DB.
This is a GET action.
All requests and responses are in JSON format. It gets informaton from an SQL
db and logs all messages from the API to a custom logger in the database.
"""
from flask import Flask, jsonify
import mysql.connector
import logging
import json
import uuid
app = Flask(__name__)
# Set up logging
logging.basicConfig(filename='api_audit.log', level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
# Database Configuration
db_config = {
'host': 'Enter IP address here',
'user': 'Enter username here',
'password': 'Enter password here',
'database': 'Enter DB name here'
}
# Function to connect to MySQL database
def connect_to_database():
try:
db_connection = mysql.connector.connect(**db_config)
return db_connection
except mysql.connector.Error as error:
logging.error("Error connecting to MySQL database: %s", error)
return None
# API Endpoint for fetching status list from status_lookup table
@app.route('/statusList', methods=['GET'])
def get_status_list():
# Log the request
request_id = str(uuid.uuid4())
logging.info("Endpoint: /statusList - Activity Type: request - Request ID: %s - Message: Request received", request_id)
# Connect to the database
db_connection = connect_to_database()
if db_connection:
cursor = db_connection.cursor(dictionary=True)
try:
# Fetch data from status_lookup table
query = "SELECT statusId, StatusDesc FROM status_lookup"
cursor.execute(query)
data = cursor.fetchall()
# Log the response
logging.info("Endpoint: /statusList - Activity Type: response - Request ID: %s - Message: Data retrieval successful", request_id)
# Close cursor and connection
cursor.close()
db_connection.close()
# Return data in JSON format
return jsonify({'status_list': data}), 200
except mysql.connector.Error as error:
logging.error("Endpoint: /statusList - Activity Type: response - Request ID: %s - Message: Data retrieval failed - Error: %s", request_id, error)
cursor.close()
db_connection.close()
return jsonify({'error': 'Data retrieval failed'}), 500
else:
return jsonify({'error': 'Database connection failed'}), 500
if __name__ == '__main__':
app.run(debug=False)