crafty-4/app/classes/models/servers.py

import logging
import datetime
from peewee import (
    CharField,
    AutoField,
    DateTimeField,
    BooleanField,
    IntegerField,
)

from app.classes.shared.main_models import DatabaseShortcuts
from app.classes.models.base_model import BaseModel

logger = logging.getLogger(__name__)

# **********************************************************************************
#                                   Servers Class
# **********************************************************************************
class Servers(BaseModel):
    server_id = AutoField()
    created = DateTimeField(default=datetime.datetime.now)
    server_uuid = CharField(default="", index=True)
    server_name = CharField(default="Server", index=True)
    path = CharField(default="")
    backup_path = CharField(default="")
    executable = CharField(default="")
    log_path = CharField(default="")
    execution_command = CharField(default="")
    auto_start = BooleanField(default=0)
    auto_start_delay = IntegerField(default=10)
    crash_detection = BooleanField(default=0)
    stop_command = CharField(default="stop")
    executable_update_url = CharField(default="")
    server_ip = CharField(default="127.0.0.1")
    server_port = IntegerField(default=25565)
    logs_delete_after = IntegerField(default=0)
    type = CharField(default="minecraft-java")

    class Meta:
        table_name = "servers"


# **********************************************************************************
#                                   Servers Class
# **********************************************************************************
class HelperServers:
    def __init__(self, database):
        self.database = database

    # **********************************************************************************
    #                                   Generic Servers Methods
    # **********************************************************************************
    @staticmethod
    def create_server(
        name: str,
        server_uuid: str,
        server_dir: str,
        backup_path: str,
        server_command: str,
        server_file: str,
        server_log_file: str,
        server_stop: str,
        server_type: str,
        server_port=25565,
    ):
        return Servers.insert(
            {
                Servers.server_name: name,
                Servers.server_uuid: server_uuid,
                Servers.path: server_dir,
                Servers.executable: server_file,
                Servers.execution_command: server_command,
                Servers.auto_start: False,
                Servers.auto_start_delay: 10,
                Servers.crash_detection: False,
                Servers.log_path: server_log_file,
                Servers.server_port: server_port,
                Servers.stop_command: server_stop,
                Servers.backup_path: backup_path,
                Servers.type: server_type,
            }
        ).execute()

    @staticmethod
    def get_server_obj(server_id):
        return Servers.get_by_id(server_id)

    @staticmethod
    def get_server_type_by_id(server_id):
        server_type = Servers.select().where(Servers.server_id == server_id).get()
        return server_type.type

    @staticmethod
    def update_server(server_obj):
        return server_obj.save()

    def remove_server(self, server_id):
        with self.database.atomic():
            Servers.delete().where(Servers.server_id == server_id).execute()

    @staticmethod
    def get_server_data_by_id(server_id):
        query = Servers.select().where(Servers.server_id == server_id).limit(1)
        try:
            return DatabaseShortcuts.return_rows(query)[0]
        except IndexError:
            return {}

    # **********************************************************************************
    #                                     Servers Methods
    # **********************************************************************************
    @staticmethod
    def get_all_defined_servers():
        query = Servers.select()
        return DatabaseShortcuts.return_rows(query)

    @staticmethod
    def get_server_friendly_name(server_id):
        server_data = HelperServers.get_server_data_by_id(server_id)
        friendly_name = (
            f"{server_data.get('server_name', None)} "
            f"with ID: {server_data.get('server_id', 0)}"
        )
        return friendly_name