Rewrote using command factory
All checks were successful
/ backend-pylint (push) Successful in 11s
/ build-artifacts (push) Successful in 10s
/ publish-artifacts (push) Successful in 8s

This commit is contained in:
Firq 2024-07-03 20:36:22 +02:00
parent 84571f1692
commit 537a0d8015
Signed by: Firq
GPG key ID: 3ACC61C8CEC83C20
12 changed files with 287 additions and 196 deletions

View file

View file

@ -0,0 +1,74 @@
[
{
"command": "host",
"description": "Sets and gets the URI of the dockge instance. Remove any unnecessary subdomains/protocols from the URI",
"args": 1,
"optional": true
},
{
"command": "login",
"description": "Logs into a given dockge account, either with an interactive dialogue or by passing --user and --password",
"args": 2,
"optional": true
},
{
"command": "logout",
"description": "Removes the credentials from the local storage.",
"args": 0,
"optional": false
},
{
"command": "list",
"description": "Lists all available stacks with their status",
"args": 0,
"optional": false
},
{
"command": "status",
"description": "Returns the status of one stack",
"args": 1,
"optional": false
},
{
"command": "restart",
"description": "Restarts a given stack",
"args": 1,
"optional": false
},
{
"command": "start",
"description": "Starts a given stack",
"args": 1,
"optional": false
},
{
"command": "stop",
"description": "Stops a given stack",
"args": 1,
"optional": false
},
{
"command": "down",
"description": "Stop & Downs a given stack",
"args": 1,
"optional": false
},
{
"command": "update",
"description": "Updates a stack",
"args": 1,
"optional": false
},
{
"command": "exit",
"description": "Exits the CLI - this will reset all settings, including credentials and host",
"args": 0,
"optional": false
},
{
"command": "help",
"description": "Displays helping hints for commands",
"args": 1,
"optional": true
}
]

View file

@ -0,0 +1,30 @@
import pathlib
import json
from typing import List, Callable
from pydantic import BaseModel
from ..components.bindings import binds
class Descriptor(BaseModel):
command: str
description: str
args: int
optional: bool
class Command(Descriptor):
binding: Callable
_descriptor_file = pathlib.Path(__file__).parent / "descriptors.json"
commands: dict[str, Command] = {}
with open(_descriptor_file, "r", encoding="utf-8") as file:
descriptors: List[Descriptor] = json.load(file)
for descriptor in descriptors:
commands.update({
descriptor["command"]:
Command(
**descriptor,
binding=binds[descriptor["command"]]
)
})

View file

@ -0,0 +1,117 @@
from urllib.parse import urlparse
from getpass import getpass
from ..models.parser import Credentials
from . import storage
from .utils import stack_formatter, status_formatter, generic_formatter, get_credential_parser
from .communicate import DockgeConnection
class ExecutionCommands():
@staticmethod
def __setup():
con = DockgeConnection()
con.connect_and_login()
return con
@staticmethod
def host(extra_args):
if len(extra_args) > 0:
res = urlparse(extra_args[0])
if all([res.scheme, res.netloc]):
host = extra_args[0].rstrip("/").replace("https://", "").replace("wss://", "")
storage.put("host", host)
else:
raise ValueError(f"Malformed URL {extra_args[0]}")
print(storage.get("host"))
@staticmethod
def login(extra_args):
if len(extra_args) > 0:
credentials = get_credential_parser().parse_args(extra_args, namespace=Credentials)
storage.put("username", credentials.username, encoded=True)
storage.put("password", credentials.password, encoded=True)
return
storage.put("username", input("Username: "), encoded=True)
storage.put("password", getpass("Password: "), encoded=True)
@staticmethod
def logout(_):
storage.remove("username")
storage.remove("password")
@staticmethod
def exit(_):
storage.clear()
@staticmethod
def list(_):
con = ExecutionCommands.__setup()
stack_formatter(con.list_stacks())
con.disconnect()
@staticmethod
def status(extra_args):
if extra_args is None:
raise ValueError
con = ExecutionCommands.__setup()
status_formatter(con.list_stack(extra_args[0]))
con.disconnect()
@staticmethod
def restart(extra_args):
if extra_args is None:
raise ValueError
con = ExecutionCommands.__setup()
generic_formatter(con.restart(extra_args[0]))
con.disconnect()
@staticmethod
def update(extra_args):
if extra_args is None:
raise ValueError
con = ExecutionCommands.__setup()
generic_formatter(con.update(extra_args[0]))
con.disconnect()
@staticmethod
def stop(extra_args):
if extra_args is None:
raise ValueError
con = ExecutionCommands.__setup()
generic_formatter(con.stop(extra_args[0]))
con.disconnect()
@staticmethod
def start(extra_args):
if extra_args is None:
raise ValueError
con = ExecutionCommands.__setup()
generic_formatter(con.start(extra_args[0]))
con.disconnect()
@staticmethod
def down(extra_args):
if extra_args is None:
raise ValueError
con = ExecutionCommands.__setup()
generic_formatter(con.down(extra_args[0]))
con.disconnect()
@staticmethod
def help():
print("WTF")
binds = {
"host": ExecutionCommands.host,
"login": ExecutionCommands.login,
"logout": ExecutionCommands.logout,
"start": ExecutionCommands.start,
"restart": ExecutionCommands.restart,
"stop": ExecutionCommands.stop,
"down": ExecutionCommands.down,
"exit": ExecutionCommands.exit,
"list": ExecutionCommands.list,
"status": ExecutionCommands.status,
"update": ExecutionCommands.update,
"help": ExecutionCommands.help,
}

View file

@ -17,6 +17,8 @@ class DockgeConnection:
def __init__(self): def __init__(self):
self._logged_in = False self._logged_in = False
self._host = storage.get("host") self._host = storage.get("host")
if self._host is None:
raise ValueError("Host for Dockge is not defined!")
self._sio = socketio.Client(logger=False, engineio_logger=False) self._sio = socketio.Client(logger=False, engineio_logger=False)
self._stacklist = None self._stacklist = None
self._init_events() self._init_events()

View file

@ -2,30 +2,17 @@ import argparse
import sys import sys
from .. import __version__ from .. import __version__
from ..models import commands from ..models.parser import Arguments
from .bindings import binds
# pylint: disable=too-few-public-methods
class Arguments(argparse.Namespace):
command: str
# pylint: disable=too-few-public-methods
class Credentials(argparse.Namespace):
username: str
password: str
credentialparser = argparse.ArgumentParser(
prog="login",
description="Subparser for login credentials provided by CI"
)
credentialparser.add_argument("--username", action="store", type=str, required=True)
credentialparser.add_argument("--password", action="store", type=str, required=True)
def parse_arguments():
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog="dockge_cli", prog="dockge_cli",
description="CLI interface for interacting with Dockge",) description="CLI interface for interacting with Dockge",)
parser.add_argument("command", choices=list(commands.keys()), action="store", type=str, default=None) parser.add_argument("command", choices=list(binds.keys()), action="store", type=str, default=None)
parser.add_argument("--version", action="version", version=f"dockge_cli {__version__}") parser.add_argument("--version", action="version", version=f"dockge_cli {__version__}")
args = Arguments() args = Arguments()
args, extra_args = parser.parse_known_args(sys.argv[1:], namespace=args) args, extra_args = parser.parse_known_args(sys.argv[1:], namespace=args)
return args, extra_args

View file

@ -1,86 +1,28 @@
from getpass import getpass from ..commands.factory import commands
from urllib.parse import urlparse
from . import storage def display_help(extra_args):
from .utils import display_help, stack_formatter, status_formatter, generic_formatter if not extra_args:
from .parser import Credentials, credentialparser print(f"{commands['help'].description}")
from .communicate import DockgeConnection
def setup():
con = DockgeConnection()
con.connect_and_login()
return con
def exec_command(command, extra_args):
match command:
case "help":
display_help(extra_args)
case "login":
if len(extra_args) > 0:
credentials = credentialparser.parse_args(extra_args, namespace=Credentials)
storage.put("username", credentials.username, encoded=True)
storage.put("password", credentials.password, encoded=True)
return return
storage.put("username", input("Username: "), encoded=True) if len(extra_args) > 1:
storage.put("password", getpass("Password: "), encoded=True) raise ValueError("Invalid arguments for help command")
case "logout": if extra_args[0] not in commands:
storage.remove("username") raise ValueError("Unknown command")
storage.remove("password") print(f"{commands[extra_args[0]].description}")
case "host":
if len(extra_args) > 0: def run(command, args):
res = urlparse(extra_args[0]) if command not in commands:
if all([res.scheme, res.netloc]): raise ValueError("Invalid Command")
host = extra_args[0].rstrip("/").replace("wss://", "").replace("https://", "")
storage.put("host", host) c = commands[command]
else:
raise ValueError(f"Malformed URL {extra_args[0]}") if args and c.args > len(args):
print(storage.get("host")) raise ValueError("Too many arguments")
case "exit": if args and c.args < len(args) and not c.optional:
storage.clear() raise ValueError("Missing arguments")
case "list":
con = setup() if command == "help":
stack_formatter(con.list_stacks()) display_help(args)
con.disconnect() return
case "status":
if extra_args is None: c.binding(args)
raise ValueError
con = setup()
status_formatter(con.list_stack(extra_args[0]))
con.disconnect()
case "restart":
if extra_args is None:
raise ValueError
con = setup()
generic_formatter(con.restart(extra_args[0]))
con.disconnect()
case "update":
if extra_args is None:
raise ValueError
con = setup()
generic_formatter(con.update(extra_args[0]))
con.disconnect()
case "stop":
if extra_args is None:
raise ValueError
con = setup()
generic_formatter(con.stop(extra_args[0]))
con.disconnect()
case "start":
if extra_args is None:
raise ValueError
con = setup()
generic_formatter(con.start(extra_args[0]))
con.disconnect()
case "down":
if extra_args is None:
raise ValueError
con = setup()
generic_formatter(con.down(extra_args[0]))
con.disconnect()
case "debug":
con = setup()
stack_formatter(con.list_stacks())
print("fgo-ta-com", con.list_stack("fgo-ta-com"))
con.disconnect()
case _:
print("Not implemented")

View file

@ -1,13 +1,15 @@
import argparse
from tabulate import tabulate from tabulate import tabulate
from ..models import commands, StackStatus from ..models import StackStatus
def display_help(extra_args): def get_credential_parser():
if not extra_args: credentialparser = argparse.ArgumentParser(
print(f"{commands['help'].description}") prog="login",
return description="Subparser for login credentials provided by CI"
if len(extra_args) > 1: )
raise ValueError("Invalid arguments for help command") credentialparser.add_argument("--username", action="store", type=str, required=True)
print(f"{commands[extra_args[0]].description}") credentialparser.add_argument("--password", action="store", type=str, required=True)
return credentialparser
def stack_formatter(stacks): def stack_formatter(stacks):
if not stacks["ok"]: if not stacks["ok"]:

View file

@ -1,5 +1,6 @@
from .components.parser import args, extra_args from .components.parser import parse_arguments
from .components.run import exec_command from .components.run import run
def cli(): def cli():
exec_command(args.command, extra_args) command, args= parse_arguments()
run(command.command, args)

View file

@ -1,2 +1 @@
from .commands import commands
from .codes import StackStatus from .codes import StackStatus

View file

@ -1,73 +0,0 @@
from pydantic import BaseModel
class CommandListing(BaseModel):
command: str
description: str
cmd_host = CommandListing(
command="host",
description="Sets and gets the URI of the dockge instance. Remove any unnecessary subdomains/protocols from the URI"
)
cmd_login = CommandListing(
command="login",
description="Logs into a given dockge account, either with an interactive dialogue or by passing --user and --password",
)
cmd_logout = CommandListing(
command="logout",
description="Removes the credentials from the local storage.",
)
cmd_list = CommandListing(
command="list",
description="Lists all available stacks with their status",
)
cmd_status = CommandListing(
command="status",
description="Returns the status of one stack",
)
cmd_restart = CommandListing(
command="restart",
description="Restarts a given stack",
)
cmd_start = CommandListing(
command="start",
description="Starts a given stack",
)
cmd_stop = CommandListing(
command="stop",
description="Stops a given stack",
)
cmd_down = CommandListing(
command="down",
description="Stop & Downs a given stack",
)
cmd_update = CommandListing(
command="update",
description="Updates a stack",
)
cmd_exit = CommandListing(
command="exit",
description="Exits the CLI - this will reset all settings, including credentials and host",
)
cmd_debug = CommandListing(
command="debug",
description="debug",
)
cmd_help = CommandListing(
command="help",
description="Displays helping hints for commands",
)
commandlist = [cmd_host, cmd_login, cmd_logout, cmd_list, cmd_restart, cmd_update, cmd_exit, cmd_debug, cmd_help, cmd_status, cmd_start, cmd_stop, cmd_down]
commands = { k.command: k for k in commandlist }

View file

@ -0,0 +1,10 @@
import argparse
# pylint: disable=too-few-public-methods
class Arguments(argparse.Namespace):
command: str
# pylint: disable=too-few-public-methods
class Credentials(argparse.Namespace):
username: str
password: str