First implementation
All checks were successful
/ backend-pylint (push) Successful in 11s

This commit is contained in:
Firq 2024-07-02 18:09:56 +02:00
parent 44ee9c7cad
commit 0c884e01cf
Signed by: Firq
GPG key ID: 3ACC61C8CEC83C20
14 changed files with 302 additions and 58 deletions

1
.gitignore vendored
View file

@ -1,6 +1,7 @@
# Environment # Environment
.vscode/ .vscode/
*venv/ *venv/
.old/
# Python stuff # Python stuff
__pycache__/ __pycache__/

View file

@ -1,4 +1,2 @@
import importlib.metadata import importlib.metadata
__version__ = importlib.metadata.version(__package__ or "dockge_cli") __version__ = importlib.metadata.version(__package__ or "dockge_cli")
from .dockge_cli import cli

View file

@ -0,0 +1,110 @@
import time
from typing import Any
import socketio
import socketio.exceptions
from . import storage
class DockgeConnection:
class LoginException(BaseException):
pass
_sio: socketio.Client
_host: str
_logged_in: bool
_stacklist: Any
def __init__(self):
self._logged_in = False
self._host = storage.get("host")
self._sio = socketio.Client(logger=False, engineio_logger=False)
self._stacklist = None
self._init_events()
def _init_events(self):
@self._sio.event
def connect():
self.connect()
print("Connected!")
@self._sio.event
def disconnect():
self._logged_in = False
print("Disconnected!")
@self._sio.event
def connect_error(data):
print("The connection failed!")
print(data)
@self._sio.on('info')
def info(data):
if all(k in data for k in ("version", "latestVersion", "isContainer", "primaryHostname")):
print(f"Dockge Version: {data['version']}")
@self._sio.on('agent')
def agent(*args):
if args[0] == "stackList":
self._stacklist = args[1]
# Callbacks
def _login_process(self, data):
success = False
if "ok" in data and "token" in data:
print(f"Login was {'successful' if data['ok'] else 'unsuccessful'}!")
success = True
else:
print("Issue with login procedure")
return success
# Functions
def connect_and_login(self):
self._sio.connect(f"https://{self._host}/socket.io/", transports=['websocket'])
self.connect()
def connect(self):
if self._logged_in:
return
data = None
retry, count = True, 0
while retry and count < 5:
try:
data = self._sio.call(
"login",
{
"username": storage.get("username", encoded=True),
"password": storage.get("password", encoded=True),
"token":""
},
timeout=5
)
retry = False
except socketio.exceptions.TimeoutError:
retry = True
count += 1
if retry or not self._login_process(data):
raise self.LoginException
self._logged_in = True
def list_stacks(self):
self._sio.emit("agent", ("", "requestStackList"))
while self._stacklist is None:
time.sleep(0.5)
retval = self._stacklist
self._stacklist = None
return retval
def list_stack(self, name: str):
ret = self._sio.call("agent", ("", "serviceStatusList", name), timeout=5)
return ret
def restart(self, name):
ret = self._sio.call("agent", ("", "restartStack", name), timeout=10)
return ret
def disconnect(self):
self._sio.emit("logout")
self._sio.disconnect()

View file

@ -1,31 +0,0 @@
from getpass import getpass
from urllib.parse import urlparse
from . import storage
def exec_command(command, extra_args):
match command:
case "login":
storage.set("username", input("Username: "), encoded=True)
storage.set("password", getpass("Password: "), encoded=True)
return
case "logout":
storage.remove("username")
storage.remove("password")
return
case "host":
if len(extra_args) > 0:
res = urlparse(extra_args[0])
if all([res.scheme, res.netloc]):
storage.set("host", extra_args[0])
else:
raise Exception(f"Malformed URL {extra_args[0]}")
return
print(storage.get("host"))
case "exit":
storage.clear()
return
case _:
print("Not implemented")
return

View file

@ -2,27 +2,30 @@ import argparse
import sys import sys
from .. import __version__ from .. import __version__
from ..models import commands
commands = [ # pylint: disable=too-few-public-methods
"host",
"login",
"logout",
"list",
"restart",
"update",
"exit"
]
class Arguments(argparse.Namespace): class Arguments(argparse.Namespace):
command: str 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)
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=commands, action="store", type=str, default=None) parser.add_argument("command", choices=list(commands.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)

View file

@ -0,0 +1,61 @@
from getpass import getpass
from urllib.parse import urlparse
from . import storage
from .utils import display_help, stack_formatter, status_formatter, restart_formatter
from .parser import Credentials, credentialparser
from .communicate import DockgeConnection
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
storage.put("username", input("Username: "), encoded=True)
storage.put("password", getpass("Password: "), encoded=True)
case "logout":
storage.remove("username")
storage.remove("password")
case "host":
if len(extra_args) > 0:
res = urlparse(extra_args[0])
if all([res.scheme, res.netloc]):
host = extra_args[0].rstrip("/").replace("wss://", "").replace("https://", "")
storage.put("host", host)
else:
raise ValueError(f"Malformed URL {extra_args[0]}")
print(storage.get("host"))
case "exit":
storage.clear()
case "list":
con = DockgeConnection()
con.connect_and_login()
stack_formatter(con.list_stacks())
con.disconnect()
case "status":
if extra_args is None:
raise ValueError
con = DockgeConnection()
con.connect_and_login()
status_formatter(con.list_stack(extra_args[0]))
con.disconnect()
case "restart":
if extra_args is None:
raise ValueError
con = DockgeConnection()
con.connect_and_login()
restart_formatter(con.restart(extra_args[0]))
con.disconnect()
case "debug":
con = DockgeConnection()
con.connect_and_login()
stack_formatter(con.list_stacks())
print("fgo-ta-com", con.list_stack("fgo-ta-com"))
con.disconnect()
case _:
print("Not implemented")

View file

@ -10,30 +10,30 @@ _storagepath.mkdir(exist_ok=True, parents=True)
def fileexists(): def fileexists():
if not _file.exists(): if not _file.exists():
with open(_file, 'a'): with open(_file, 'a', encoding="utf-8"):
os.utime(_file, None) os.utime(_file, None)
def set(key: str, value: str, encoded=False): def put(key: str, value: str, encoded=False):
fileexists() fileexists()
with open(_file, "r+") as file: with open(_file, "r+", encoding="utf-8") as file:
content: dict[str, str] = yaml.load(file, Loader=yaml.SafeLoader) or {} content: dict[str, str] = yaml.load(file, Loader=yaml.SafeLoader) or {}
content.update({ key: base64.b64encode(value.encode()) if encoded else value }) content.update({ key: base64.b64encode(value.encode()) if encoded else value })
with open(_file, "w+") as file: with open(_file, "w+", encoding="utf-8") as file:
yaml.dump(content, file, Dumper=yaml.SafeDumper) yaml.dump(content, file, Dumper=yaml.SafeDumper)
def remove(key: str): def remove(key: str):
fileexists() fileexists()
with open(_file, "r") as file: with open(_file, "r", encoding="utf-8") as file:
content: dict[str, str] = yaml.load(file, Loader=yaml.SafeLoader) or {} content: dict[str, str] = yaml.load(file, Loader=yaml.SafeLoader) or {}
content.pop(key, None) content.pop(key, None)
with open(_file, "w+") as file: with open(_file, "w+", encoding="utf-8") as file:
yaml.dump(content, file, Dumper=yaml.SafeDumper) yaml.dump(content, file, Dumper=yaml.SafeDumper)
def get(key: str, encoded=False): def get(key: str, encoded=False):
value: str = None value: str = None
if not _file.exists(): if not _file.exists():
return None return None
with open(_file, "r") as file: with open(_file, "r", encoding="utf-8") as file:
content: dict[str, str] = yaml.load(file, Loader=yaml.SafeLoader) content: dict[str, str] = yaml.load(file, Loader=yaml.SafeLoader)
value = content.get(key, None) value = content.get(key, None)
if value is None: if value is None:

View file

@ -0,0 +1,30 @@
from tabulate import tabulate
from ..models import commands, StackStatus
def display_help(extra_args):
if not extra_args:
print(f"{commands['help'].description}")
return
if len(extra_args) > 1:
raise ValueError("Invalid arguments for help command")
print(f"{commands[extra_args[0]].description}")
def stack_formatter(stacks):
if not stacks["ok"]:
raise RuntimeError("Stack GET didn't work")
table, headers = [], ["Stackname", "Status"]
for key, val in stacks["stackList"].items():
table.append([key, StackStatus(val['status']).name])
print(tabulate(table, headers=headers, tablefmt="github"), "\n")
def status_formatter(status):
print(f"Is Stack Ok? {'Yes' if status['ok'] else 'No'}")
headers = ["Container", "Status"]
table = [[k, v] for k, v in status["serviceStatusList"].items()]
print(tabulate(table, headers=headers, tablefmt="github"), "\n")
def restart_formatter(status):
print(f"Is Ok? {'Yes' if status['ok'] else 'No'}")
print(f"Stack status: {status['msg']}")

View file

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

View file

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

View file

@ -0,0 +1,7 @@
from enum import Enum
class StackStatus(Enum):
# pylint: disable=invalid-name
inactive = 1
running = 3
exited = 4

View file

@ -0,0 +1,58 @@
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_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]
commands = { k.command: k for k in commandlist }

View file

@ -2,7 +2,11 @@
name = "dockge_cli" name = "dockge_cli"
version = "0.0.1-a.1" version = "0.0.1-a.1"
dependencies = [ dependencies = [
"pyyaml~=6.0.1" "pyyaml~=6.0.1",
"pydantic~=2.8.0",
"python-socketio~=5.11.3",
"websocket-client~=1.8.0",
"tabulate ~=0.9.0",
] ]
requires-python = ">= 3.10" requires-python = ">= 3.10"
authors = [{name = "Firq", email = "firelp42@gmail.com"}] authors = [{name = "Firq", email = "firelp42@gmail.com"}]
@ -17,14 +21,15 @@ classifiers = [
] ]
[project.scripts] [project.scripts]
dockge-cli = "dockge_cli:cli" dockge-cli = "dockge_cli.dockge_cli:cli"
dockge = "dockge_cli.dockge_cli:cli"
[tool.setuptools.packages.find] [tool.setuptools.packages.find]
where = ["."] where = ["."]
include = ["dockge_cli*"] include = ["dockge_cli*"]
[tool.pylint."MAIN"] [tool.pylint."MAIN"]
disable = [ "line-too-long", "missing-module-docstring" ] disable = [ "line-too-long", "missing-module-docstring", "missing-function-docstring", "missing-class-docstring" ]
[build-system] [build-system]
requires = ["setuptools >= 61.0"] requires = ["setuptools >= 61.0"]