mirror of
https://github.com/hubHarmony/servii-backend.git
synced 2024-11-17 21:40:31 +00:00
87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
import http
|
|
import inspect
|
|
|
|
from flask import Flask, Response, jsonify, request
|
|
from flask_cors import CORS
|
|
|
|
import generic_executor
|
|
|
|
app = Flask(__name__)
|
|
CORS(app)
|
|
cors = CORS(app, origins=['*'])
|
|
|
|
|
|
def generic_response_maker(status_code: http.HTTPStatus, _message: str = None) -> tuple[Response, int]:
|
|
if _message is not None:
|
|
return jsonify({'message': _message}), status_code.value
|
|
match status_code:
|
|
case http.HTTPStatus.CREATED:
|
|
message = jsonify({'message': 'Creation successful.'})
|
|
case http.HTTPStatus.INTERNAL_SERVER_ERROR:
|
|
message = jsonify({'message': 'Internal Server Error.'})
|
|
case http.HTTPStatus.NO_CONTENT:
|
|
message = jsonify({'message': 'Deletion successful.'})
|
|
case http.HTTPStatus.ACCEPTED:
|
|
message = jsonify({'message': 'Action successful.'})
|
|
case http.HTTPStatus.BAD_REQUEST:
|
|
message = jsonify({'message': 'Bad Request, the property you tried to modify is not valid.'})
|
|
case http.HTTPStatus.NOT_FOUND:
|
|
message = jsonify({'message': 'Server not found.'})
|
|
case http.HTTPStatus.UNSUPPORTED_MEDIA_TYPE:
|
|
message = jsonify({'message': 'Unsupported Media Type / No JSON payload'})
|
|
case http.HTTPStatus.OK:
|
|
message = jsonify({'message': 'Success.'})
|
|
case http.HTTPStatus.METHOD_NOT_ALLOWED:
|
|
message = jsonify({'message': 'This API call does not exist.'})
|
|
case _:
|
|
status_code = http.HTTPStatus.BAD_GATEWAY
|
|
message = jsonify({'message': 'Bad Gateway.'})
|
|
return message, status_code.value
|
|
|
|
|
|
def parse_and_validate_request(parameters: [str]) -> list[str]:
|
|
args = []
|
|
data = request.get_json()
|
|
if not data:
|
|
raise Exception("Empty request body.")
|
|
for name in parameters:
|
|
if name not in data:
|
|
raise Exception(f"Missing parameter {name}")
|
|
value = data[name]
|
|
if isinstance(value, str):
|
|
args.append(value)
|
|
return args
|
|
|
|
|
|
route_handlers = {
|
|
'AccountCreate': generic_executor.account_create,
|
|
'ServerCreate': generic_executor.server_create,
|
|
'ServerDelete': generic_executor.server_delete,
|
|
'AccountDelete': generic_executor.account_delete,
|
|
'ServerRun': generic_executor.server_run,
|
|
'ServerStop': generic_executor.server_stop,
|
|
'UpdateProperty': generic_executor.update_property,
|
|
'Command': generic_executor.run_command,
|
|
}
|
|
|
|
|
|
@app.route('/<path:path>', methods=['POST'])
|
|
def dynamic_route_handler(path):
|
|
if path not in route_handlers:
|
|
return generic_response_maker(http.HTTPStatus.METHOD_NOT_ALLOWED)
|
|
|
|
route_fn = route_handlers[path]
|
|
parameters = []
|
|
sig = inspect.signature(route_fn)
|
|
|
|
for param in sig.parameters.values():
|
|
parameters.append(param.name)
|
|
|
|
mapped_parameters = parse_and_validate_request(parameters)
|
|
status: http.HTTPStatus = route_fn(*mapped_parameters)
|
|
return generic_response_maker(status)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=3005, debug=True)
|