[+] Routes handling refactor | from charleslemaux/ROUTES-UPDATE

[+] Routes handling refactor

api.py

@@ -1,6 +1,5 @@
 import http
-from enum import Enum
+import inspect
-from typing import Dict, Optional
 
 from flask import Flask, Response, jsonify, request
 from flask_cors import CORS
@@ -12,12 +11,9 @@ CORS(app)
 cors = CORS(app, origins=['*'])
 
 
-class ActionType(Enum):
+def generic_response_maker(status_code: http.HTTPStatus, _message: str = None) -> tuple[Response, int]:
-    ACCOUNT_CREATE = '/AccountCreate'
+    if _message is not None:
-    SERVER_CREATE = '/ServerCreate'
+        return jsonify({'message': _message}), status_code.value
-
-
-def generic_response_maker(status_code: http.HTTPStatus) -> tuple[Response, int]:
     match status_code:
         case http.HTTPStatus.CREATED:
             message = jsonify({'message': 'Creation successful.'})
@@ -35,133 +31,54 @@ def generic_response_maker(status_code: http.HTTPStatus) -> tuple[Response, int]
             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(required_keys: set) -> Optional[dict]:
+def parse_and_validate_request(parameters: [str]) -> list[str]:
+    args = []
     data = request.get_json()
     if not data:
         raise Exception("Empty request body.")
-    missing_keys = required_keys - set(data.keys())
+    for name in parameters:
-    if missing_keys:
+        if name not in data:
-        raise Exception("Missing keys: {}".format(missing_keys))
+            raise Exception(f"Missing parameter {name}")
-    return data
+        value = data[name]
+        if isinstance(value, str):
+            args.append(value)
+    return args
 
 
-@app.route('/AccountCreate', methods=['POST'])
+route_handlers = {
-def account_create() -> tuple[Response, int]:
+    'AccountCreate': generic_executor.account_create,
-    data: Dict[str, str] = request.get_json()
+    'ServerCreate': generic_executor.server_create,
-    if not data:
+    'ServerDelete': generic_executor.server_delete,
-        return generic_response_maker(http.HTTPStatus.UNSUPPORTED_MEDIA_TYPE)
+    'AccountDelete': generic_executor.account_delete,
-
+    'ServerRun': generic_executor.server_run,
-    email: str = data.get('email')
+    'ServerStop': generic_executor.server_stop,
-    port: str = data.get('port')
+    'UpdateProperty': generic_executor.update_property,
-    print(f'Email received: {email}')
+    'Command': generic_executor.run_command,
-    print(f'Port received: {port}')
+}
-    status: http.HTTPStatus = generic_executor.account_create(port)
-    return generic_response_maker(status)
 
 
-@app.route('/ServerCreate', methods=['POST'])
+@app.route('/<path:path>', methods=['POST'])
-def server_create() -> tuple[Response, int]:
+def dynamic_route_handler(path):
-    data: Dict[str, str] = request.get_json()
+    if path not in route_handlers:
-    if not data:
+        return generic_response_maker(http.HTTPStatus.METHOD_NOT_ALLOWED)
-        return generic_response_maker(http.HTTPStatus.UNSUPPORTED_MEDIA_TYPE)
 
-    port: str = data.get('port')
+    route_fn = route_handlers[path]
-    name: str = data.get('name')
+    parameters = []
-    version: str = data.get('version')
+    sig = inspect.signature(route_fn)
-    print(f'Port received for server creation: {port}')
-    print(f'Server name: {name}')
-    print(f'Server version: {version}')
-    status: http.HTTPStatus = generic_executor.server_create(port, name, version)
-    return generic_response_maker(status)
 
+    for param in sig.parameters.values():
+        parameters.append(param.name)
 
-@app.route('/ServerDelete', methods=['POST'])
+    mapped_parameters = parse_and_validate_request(parameters)
-def server_delete() -> tuple[Response, int]:
+    status: http.HTTPStatus = route_fn(*mapped_parameters)
-    data: Dict[str, str] = request.get_json()
-    if not data:
-        return generic_response_maker(http.HTTPStatus.UNSUPPORTED_MEDIA_TYPE)
-
-    port: str = data.get('port')
-    name: str = data.get('name')
-    print(f'Port received for server deletion: {port}')
-    print(f'Server name to delete: {name}')
-    status: http.HTTPStatus = generic_executor.server_delete(port, name)
-    return generic_response_maker(status)
-
-
-@app.route('/AccountDelete', methods=['POST'])
-def account_delete() -> tuple[Response, int]:
-    data: Dict[str, str] = request.get_json()
-    if not data:
-        return generic_response_maker(http.HTTPStatus.UNSUPPORTED_MEDIA_TYPE)
-
-    port: str = data.get('port')
-    email: str = data.get('email')
-    print(f'Port received for account deletion: {port}')
-    print(f'Email of account to delete: {email}')
-    status: http.HTTPStatus = generic_executor.account_delete(port)
-    return generic_response_maker(status)
-
-
-@app.route('/ServerRun', methods=['POST'])
-def server_run() -> tuple[Response, int]:
-    data: Dict[str, str] = request.get_json()
-    if not data:
-        return generic_response_maker(http.HTTPStatus.UNSUPPORTED_MEDIA_TYPE)
-
-    port: str = data.get('port')
-    name: str = data.get('name')
-    print(f'Port received for server launch: {port}')
-    print(f'Server name to launch: {name}')
-    status: http.HTTPStatus = generic_executor.server_run(port, name)
-    return generic_response_maker(status)
-
-
-@app.route('/ServerStop', methods=['POST'])
-def server_stop() -> tuple[Response, int]:
-    data: Dict[str, str] = request.get_json()
-    if not data:
-        return generic_response_maker(http.HTTPStatus.UNSUPPORTED_MEDIA_TYPE)
-
-    port: str = data.get('port')
-    name: str = data.get('name')
-    print(f'Port received for server shutdown: {port}')
-    print(f'Server name to shut down: {name}')
-    status: http.HTTPStatus = generic_executor.server_stop(port, name)
-    return generic_response_maker(status)
-
-
-@app.route('/UpdateProperty', methods=['POST'])
-def update_property() -> tuple[Response, int]:
-    data: Dict[str, str, str, str] = request.get_json()
-    if not data:
-        return generic_response_maker(http.HTTPStatus.UNSUPPORTED_MEDIA_TYPE)
-
-    port: str = data.get('port')
-    name: str = data.get('name')
-    prop: str = data.get('property')
-    value: str = data.get('value')
-    status: http.HTTPStatus = generic_executor.update_property(port, name, prop, value)
-    print(f'The server {name} at port {port} changed {prop} to {value}')
-    return generic_response_maker(status)
-
-
-@app.route('/Command', methods=['POST'])
-def command() -> tuple[Response, int]:
-    data: Dict[str, str, str] = request.get_json()
-    if not data:
-        return generic_response_maker(http.HTTPStatus.UNSUPPORTED_MEDIA_TYPE)
-
-    port: str = data.get('port')
-    _command: str = data.get('command')
-    print(f'Server {port} executed command {_command}')
-    status: http.HTTPStatus = generic_executor.run_command(port, _command)
     return generic_response_maker(status)
 
 

api_sender_v1.html

@@ -46,7 +46,7 @@ document.addEventListener('DOMContentLoaded', () => {
             const port = document.getElementById('accountPort').value;
             const name = document.getElementById('serverName').value;
             const version = document.getElementById('serverVersion').value;
-            const property = document.getElementById('update_property').value;
+            const prop = document.getElementById('update_property').value;
             const value = document.getElementById('update_value').value;
             const command = document.getElementById('command').value;
             var data = {}
@@ -70,7 +70,7 @@ document.addEventListener('DOMContentLoaded', () => {
                     data = {port, name}
                     break;
                 case 'UpdateProperty':
-                    data = {port, name, property, value}
+                    data = {port, name, prop, value}
                     break;
                 case 'Command':
                     data = {port, name, command}

server_mc_manager.py

@@ -23,7 +23,8 @@ class MinecraftServerManager:
     def execute_server_command(self, server_id, command):
         if server_id in self.servers:
             process = self.servers[server_id]['process']
-            process.stdin.write(command.encode() + b'\n')
+            process.stdin.write(command.encode('utf-8') + b'\n')
+            process.stdin.flush()
             print(f"Server {server_id} executed command : {command}")
         else:
             print(f"No server found with ID {server_id}")