Improved typing, refactored code, fixed some issues

dockge_cli/service/communicate.py

@@ -0,0 +1,128 @@
+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")
+        if self._host is None:
+            raise ValueError("Host for Dockge is not defined!")
+        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 update(self, name):
+        ret = self._sio.call("agent", ("", "updateStack", name), timeout=10)
+        return ret
+
+    def stop(self, name):
+        ret = self._sio.call("agent", ("", "stopStack", name), timeout=10)
+        return ret
+
+    def start(self, name):
+        ret = self._sio.call("agent", ("", "startStack", name), timeout=10)
+        return ret
+
+    def down(self, name):
+        ret = self._sio.call("agent", ("", "downStack", name), timeout=10)
+        return ret
+
+    def disconnect(self):
+        self._sio.emit("logout")
+        self._sio.disconnect()

dockge_cli/service/storage.py

@@ -0,0 +1,44 @@
+import os
+import pathlib
+import base64
+import yaml
+
+_storagepath = pathlib.Path(__file__).parents[1] / ".temp"
+_file = _storagepath / "storage.yaml"
+
+_storagepath.mkdir(exist_ok=True, parents=True)
+
+def fileexists():
+    if not _file.exists():
+        with open(_file, 'a', encoding="utf-8"):
+            os.utime(_file, None)
+
+def put(key: str, value: str, encoded=False):
+    fileexists()
+    with open(_file, "r+", encoding="utf-8") as file:
+        content: dict[str, str] = yaml.load(file, Loader=yaml.SafeLoader) or {}
+        content.update({ key: str(base64.b64encode(value.encode())) if encoded else value })
+    with open(_file, "w+", encoding="utf-8") as file:
+        yaml.dump(content, file, Dumper=yaml.SafeDumper)
+
+def remove(key: str):
+    fileexists()
+    with open(_file, "r", encoding="utf-8") as file:
+        content: dict[str, str] = yaml.load(file, Loader=yaml.SafeLoader) or {}
+        content.pop(key, None)
+    with open(_file, "w+", encoding="utf-8") as file:
+        yaml.dump(content, file, Dumper=yaml.SafeDumper)
+
+def get(key: str, encoded=False):
+    value: str | None = None
+    if not _file.exists():
+        return None
+    with open(_file, "r", encoding="utf-8") as file:
+        content: dict[str, str] = yaml.load(file, Loader=yaml.SafeLoader)
+        value = content.get(key, None)
+    if value is None:
+        return None
+    return base64.b64decode(value).decode() if encoded else value
+
+def clear():
+    _file.unlink()