Pylint settings
All checks were successful
/ pylint (push) Successful in 11s
/ mypy (push) Successful in 13s

This commit is contained in:
Firq 2024-07-05 16:05:35 +02:00
parent be932ea9c4
commit a99b81b0c7
Signed by: Firq
GPG key ID: 3ACC61C8CEC83C20
12 changed files with 140 additions and 3 deletions

2
dockge_cli/__main__.py Normal file
View file

@ -0,0 +1,2 @@
from .dockge_cli import cli
cli()

View file

@ -7,6 +7,10 @@ from ...service.communicate import DockgeConnection
from ..utils import stack_formatter, status_formatter, generic_formatter, get_credential_parser from ..utils import stack_formatter, status_formatter, generic_formatter, get_credential_parser
class ExecutionCommands(): class ExecutionCommands():
"""
Helper class that provides all the static methods in an organized way
This is an abstraction layer of the CLI, as those functions only do little preprocessing before calling the actural DockgeConnection
"""
@staticmethod @staticmethod
def __setup(): def __setup():
con = DockgeConnection() con = DockgeConnection()
@ -15,6 +19,9 @@ class ExecutionCommands():
@staticmethod @staticmethod
def host(extra_args): def host(extra_args):
"""
host command binding
"""
if len(extra_args) > 0: if len(extra_args) > 0:
res = urlparse(extra_args[0]) res = urlparse(extra_args[0])
if all([res.scheme, res.netloc]): if all([res.scheme, res.netloc]):
@ -26,6 +33,9 @@ class ExecutionCommands():
@staticmethod @staticmethod
def login(extra_args): def login(extra_args):
"""
login command binding
"""
if len(extra_args) > 0: if len(extra_args) > 0:
credentials = get_credential_parser().parse_args(extra_args, namespace=Credentials) credentials = get_credential_parser().parse_args(extra_args, namespace=Credentials)
storage.put("username", credentials.username, encoded=True) storage.put("username", credentials.username, encoded=True)
@ -36,21 +46,33 @@ class ExecutionCommands():
@staticmethod @staticmethod
def logout(_): def logout(_):
"""
logout command binding
"""
storage.remove("username") storage.remove("username")
storage.remove("password") storage.remove("password")
@staticmethod @staticmethod
def exit(_): def exit(_):
"""
exit command binding
"""
storage.clear() storage.clear()
@staticmethod @staticmethod
def list(_): def list(_):
"""
list command binding
"""
con = ExecutionCommands.__setup() con = ExecutionCommands.__setup()
stack_formatter(con.list_stacks()) stack_formatter(con.list_stacks())
con.disconnect() con.disconnect()
@staticmethod @staticmethod
def status(extra_args): def status(extra_args):
"""
status command binding
"""
if extra_args is None: if extra_args is None:
raise ValueError raise ValueError
con = ExecutionCommands.__setup() con = ExecutionCommands.__setup()
@ -59,6 +81,9 @@ class ExecutionCommands():
@staticmethod @staticmethod
def restart(extra_args): def restart(extra_args):
"""
restart command binding
"""
if extra_args is None: if extra_args is None:
raise ValueError raise ValueError
con = ExecutionCommands.__setup() con = ExecutionCommands.__setup()
@ -67,6 +92,9 @@ class ExecutionCommands():
@staticmethod @staticmethod
def update(extra_args): def update(extra_args):
"""
update command binding
"""
if extra_args is None: if extra_args is None:
raise ValueError raise ValueError
con = ExecutionCommands.__setup() con = ExecutionCommands.__setup()
@ -75,6 +103,9 @@ class ExecutionCommands():
@staticmethod @staticmethod
def stop(extra_args): def stop(extra_args):
"""
stop command binding
"""
if extra_args is None: if extra_args is None:
raise ValueError raise ValueError
con = ExecutionCommands.__setup() con = ExecutionCommands.__setup()
@ -83,6 +114,9 @@ class ExecutionCommands():
@staticmethod @staticmethod
def start(extra_args): def start(extra_args):
"""
start command binding
"""
if extra_args is None: if extra_args is None:
raise ValueError raise ValueError
con = ExecutionCommands.__setup() con = ExecutionCommands.__setup()
@ -91,6 +125,9 @@ class ExecutionCommands():
@staticmethod @staticmethod
def down(extra_args): def down(extra_args):
"""
down command binding
"""
if extra_args is None: if extra_args is None:
raise ValueError raise ValueError
con = ExecutionCommands.__setup() con = ExecutionCommands.__setup()
@ -99,4 +136,7 @@ class ExecutionCommands():
@staticmethod @staticmethod
def help(): def help():
"""
exit command binding - This should never be invoked
"""
print("WTF") print("WTF")

View file

@ -6,6 +6,10 @@ from ..models import Arguments
from .commandprovider.factory import commands from .commandprovider.factory import commands
def parse_arguments(): def parse_arguments():
"""
Create a parser and parse the arguments of sys.argv based on the Arguments Namespace
Returns arguments and extra arguments separately
"""
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",)

View file

@ -1,6 +1,9 @@
from .commandprovider.factory import commands from .commandprovider.factory import commands
def display_help(extra_args): def display_help(extra_args):
"""
Display help dialogues for each command
"""
if not extra_args: if not extra_args:
print(f"{commands['help'].description}") print(f"{commands['help'].description}")
return return
@ -11,6 +14,10 @@ def display_help(extra_args):
print(f"{commands[extra_args[0]].description}") print(f"{commands[extra_args[0]].description}")
def run(command, args): def run(command, args):
"""
Runs a given command with the provided args
Alsso automatically maps the given command string to the correct Command class
"""
if command not in commands: if command not in commands:
raise ValueError("Invalid Command") raise ValueError("Invalid Command")
@ -25,4 +32,4 @@ def run(command, args):
display_help(args) display_help(args)
return return
c.binding(args) c.bind(args)

View file

@ -3,6 +3,9 @@ from tabulate import tabulate
from ..models import StackStatus from ..models import StackStatus
def get_credential_parser(): def get_credential_parser():
"""
Creates a new parser for login credentials
"""
credentialparser = argparse.ArgumentParser( credentialparser = argparse.ArgumentParser(
prog="login", prog="login",
description="Subparser for login credentials provided by CI" description="Subparser for login credentials provided by CI"
@ -12,6 +15,9 @@ def get_credential_parser():
return credentialparser return credentialparser
def stack_formatter(stacks): def stack_formatter(stacks):
"""
Prints a given stack list formatted as a table
"""
if not stacks["ok"]: if not stacks["ok"]:
raise RuntimeError("Stack GET didn't work") raise RuntimeError("Stack GET didn't work")
@ -22,11 +28,17 @@ def stack_formatter(stacks):
print(tabulate(table, headers=headers, tablefmt="github"), "\n") print(tabulate(table, headers=headers, tablefmt="github"), "\n")
def status_formatter(status): def status_formatter(status):
"""
Prints the status for a given stack
"""
print(f"Is Stack Ok? {'Yes' if status['ok'] else 'No'}") print(f"Is Stack Ok? {'Yes' if status['ok'] else 'No'}")
headers = ["Container", "Status"] headers = ["Container", "Status"]
table = [[k, v] for k, v in status["serviceStatusList"].items()] table = [[k, v] for k, v in status["serviceStatusList"].items()]
print(tabulate(table, headers=headers, tablefmt="github"), "\n") print(tabulate(table, headers=headers, tablefmt="github"), "\n")
def generic_formatter(status): def generic_formatter(status):
"""
Prints a generic dockge message
"""
print(f"Is Ok? {'Yes' if status['ok'] else 'No'}") print(f"Is Ok? {'Yes' if status['ok'] else 'No'}")
print(f"Stack status: {status['msg']}") print(f"Stack status: {status['msg']}")

View file

@ -2,5 +2,8 @@ from .client.parser import parse_arguments
from .client.run import run from .client.run import run
def cli(): def cli():
"""
main function for cli invocation
"""
command, args= parse_arguments() command, args= parse_arguments()
run(command.command, args) run(command.command, args)

View file

@ -1,6 +1,9 @@
from enum import Enum from enum import Enum
class StackStatus(Enum): class StackStatus(Enum):
"""
mapping codes for status vs text
"""
# pylint: disable=invalid-name # pylint: disable=invalid-name
inactive = 1 inactive = 1
running = 3 running = 3

View file

@ -2,6 +2,9 @@ from typing import Callable
from pydantic import BaseModel from pydantic import BaseModel
class Command(BaseModel): class Command(BaseModel):
"""
Basic command structure for the CLI to automatically generate valid commands
"""
cmd: str cmd: str
bind: Callable bind: Callable
args: int args: int

View file

@ -2,9 +2,15 @@ import argparse
# pylint: disable=too-few-public-methods # pylint: disable=too-few-public-methods
class Arguments(argparse.Namespace): class Arguments(argparse.Namespace):
"""
Default Arguments when calling the CLI
"""
command: str command: str
# pylint: disable=too-few-public-methods # pylint: disable=too-few-public-methods
class Credentials(argparse.Namespace): class Credentials(argparse.Namespace):
"""
Special Argument Namespace for login credentials of the login commands
"""
username: str username: str
password: str password: str

View file

@ -5,8 +5,14 @@ import socketio.exceptions
from . import storage from . import storage
class DockgeConnection: class DockgeConnection:
"""
Provider class for Dockge
Provides all the functionality for connecting, logging in and executing commands
"""
class LoginException(BaseException): class LoginException(BaseException):
pass """
Special exception when login fails too often
"""
_sio: socketio.Client _sio: socketio.Client
_host: str _host: str
@ -61,10 +67,17 @@ class DockgeConnection:
# Functions # Functions
def connect_and_login(self): def connect_and_login(self):
"""
Connect to the websocket
"""
self._sio.connect(f"https://{self._host}/socket.io/", transports=['websocket']) self._sio.connect(f"https://{self._host}/socket.io/", transports=['websocket'])
self.connect() self.connect()
def connect(self): def connect(self):
"""
Log into dockge using basicauth
Retries 5 times when timeouts occur
"""
if self._logged_in: if self._logged_in:
return return
@ -92,6 +105,9 @@ class DockgeConnection:
self._logged_in = True self._logged_in = True
def list_stacks(self): def list_stacks(self):
"""
Requests stack list from dockge, returns list when event was sent
"""
self._sio.emit("agent", ("", "requestStackList")) self._sio.emit("agent", ("", "requestStackList"))
while self._stacklist is None: while self._stacklist is None:
time.sleep(0.5) time.sleep(0.5)
@ -100,29 +116,50 @@ class DockgeConnection:
return retval return retval
def list_stack(self, name: str): def list_stack(self, name: str):
"""
Lists status for a stack
"""
ret = self._sio.call("agent", ("", "serviceStatusList", name), timeout=5) ret = self._sio.call("agent", ("", "serviceStatusList", name), timeout=5)
return ret return ret
def restart(self, name): def restart(self, name):
"""
Restarts a given stack
"""
ret = self._sio.call("agent", ("", "restartStack", name), timeout=10) ret = self._sio.call("agent", ("", "restartStack", name), timeout=10)
return ret return ret
def update(self, name): def update(self, name):
"""
Updates a given stack
"""
ret = self._sio.call("agent", ("", "updateStack", name), timeout=10) ret = self._sio.call("agent", ("", "updateStack", name), timeout=10)
return ret return ret
def stop(self, name): def stop(self, name):
"""
Stops a given stack
"""
ret = self._sio.call("agent", ("", "stopStack", name), timeout=10) ret = self._sio.call("agent", ("", "stopStack", name), timeout=10)
return ret return ret
def start(self, name): def start(self, name):
"""
Starts a given stack
"""
ret = self._sio.call("agent", ("", "startStack", name), timeout=10) ret = self._sio.call("agent", ("", "startStack", name), timeout=10)
return ret return ret
def down(self, name): def down(self, name):
"""
Stops and downs a given stack
"""
ret = self._sio.call("agent", ("", "downStack", name), timeout=10) ret = self._sio.call("agent", ("", "downStack", name), timeout=10)
return ret return ret
def disconnect(self): def disconnect(self):
"""
Logs out of dockge
"""
self._sio.emit("logout") self._sio.emit("logout")
self._sio.disconnect() self._sio.disconnect()

View file

@ -9,11 +9,18 @@ _file = _storagepath / "storage.yaml"
_storagepath.mkdir(exist_ok=True, parents=True) _storagepath.mkdir(exist_ok=True, parents=True)
def fileexists(): def fileexists():
"""
Checks if storage file does exist, creates it when necessary
"""
if not _file.exists(): if not _file.exists():
with open(_file, 'a', encoding="utf-8"): with open(_file, 'a', encoding="utf-8"):
os.utime(_file, None) os.utime(_file, None)
def put(key: str, value: str, encoded=False): def put(key: str, value: str, encoded=False):
"""
Puts a given value with a given key into the storage file
Encodes the data as base64 when encoded is set to true
"""
fileexists() fileexists()
with open(_file, "r+", encoding="utf-8") 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 {}
@ -22,6 +29,9 @@ def put(key: str, value: str, encoded=False):
yaml.dump(content, file, Dumper=yaml.SafeDumper) yaml.dump(content, file, Dumper=yaml.SafeDumper)
def remove(key: str): def remove(key: str):
"""
Removed a given key from the storage file
"""
fileexists() fileexists()
with open(_file, "r", encoding="utf-8") 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 {}
@ -30,6 +40,10 @@ def remove(key: str):
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):
"""
Retrieves a value for a given key from the storage file
If the value was encoded, encoded needs to be set True to decode it again
"""
value: str | None = None value: str | None = None
if not _file.exists(): if not _file.exists():
return None return None
@ -41,4 +55,7 @@ def get(key: str, encoded=False):
return base64.b64decode(value).decode() if encoded else value return base64.b64decode(value).decode() if encoded else value
def clear(): def clear():
"""
Deletes the storage file
"""
_file.unlink() _file.unlink()

View file

@ -42,7 +42,10 @@ typing = [
] ]
[tool.pylint."MAIN"] [tool.pylint."MAIN"]
disable = [ "line-too-long", "missing-module-docstring", "missing-function-docstring", "missing-class-docstring" ] disable = [
"line-too-long",
"missing-module-docstring",
]
[tool.mypy] [tool.mypy]
python_version = "3.11" python_version = "3.11"