servii-backend/api.py

import http
from enum import Enum
from typing import Dict, Optional

from flask import Flask, Response, jsonify, request
from flask_cors import CORS

import generic_executor

app = Flask(__name__)
CORS(app)
cors = CORS(app, origins=['*'])


class ActionType(Enum):
    ACCOUNT_CREATE = '/AccountCreate'
    SERVER_CREATE = '/ServerCreate'


def generic_response_maker(status_code: http.HTTPStatus) -> tuple[Response, int]:
    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 _:
            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]:
    data = request.get_json()
    if not data:
        raise Exception("Empty request body.")
    missing_keys = required_keys - set(data.keys())
    if missing_keys:
        raise Exception("Missing keys: {}".format(missing_keys))
    return data


@app.route('/AccountCreate', methods=['POST'])
def account_create() -> tuple[Response, int]:
    data: Dict[str, str] = request.get_json()
    if not data:
        return generic_response_maker(http.HTTPStatus.UNSUPPORTED_MEDIA_TYPE)

    email: str = data.get('email')
    port: str = data.get('port')
    print(f'Email received: {email}')
    print(f'Port received: {port}')
    status: http.HTTPStatus = generic_executor.account_create(port)
    return generic_response_maker(status)


@app.route('/ServerCreate', methods=['POST'])
def server_create() -> 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')
    version: str = data.get('version')
    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)


@app.route('/ServerDelete', methods=['POST'])
def server_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')
    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)


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3005, debug=True)