Compare commits

..

No commits in common. "main" and "0.0.1-c.1" have entirely different histories.

25 changed files with 193 additions and 494 deletions

View file

@ -6,7 +6,7 @@ on:
- '[0-9]+\.[0-9]+\.[0-9]' - '[0-9]+\.[0-9]+\.[0-9]'
jobs: jobs:
lint-and-typing: backend-pylint:
runs-on: docker runs-on: docker
container: nikolaik/python-nodejs:python3.11-nodejs21 container: nikolaik/python-nodejs:python3.11-nodejs21
steps: steps:
@ -14,19 +14,16 @@ jobs:
uses: https://code.forgejo.org/actions/checkout@v3 uses: https://code.forgejo.org/actions/checkout@v3
- name: Install packages - name: Install packages
run: | run: |
pip install -e .[lint,typing] -q --disable-pip-version-check -q pip install -e . -q
python -m pip list --format=columns --disable-pip-version-check python -m pip list --format=columns --disable-pip-version-check
pip install pylint~=2.17.7 --disable-pip-version-check -q
- name: Run pylint - name: Run pylint
run: | run: |
pylint --version pylint --version
pylint **/*.py --exit-zero --rc-file pyproject.toml pylint **/*.py --exit-zero --rc-file pyproject.toml
- name: Run mypy
run: |
mypy --version
mypy .
build-artifacts: build-artifacts:
needs: ["lint-and-typing"] needs: ["backend-pylint"]
runs-on: docker runs-on: docker
container: nikolaik/python-nodejs:python3.11-nodejs21 container: nikolaik/python-nodejs:python3.11-nodejs21
steps: steps:
@ -56,37 +53,3 @@ jobs:
run: pip install twine run: pip install twine
- name: Upload package to registry - name: Upload package to registry
run: python -m twine upload --repository-url ${{ secrets.REPOSITORY_URL }} -u ${{ secrets.TWINE_DEPLOY_USER }} -p ${{ secrets.TWINE_DEPLOY_PASSWORD }} dist/* run: python -m twine upload --repository-url ${{ secrets.REPOSITORY_URL }} -u ${{ secrets.TWINE_DEPLOY_USER }} -p ${{ secrets.TWINE_DEPLOY_PASSWORD }} dist/*
build-and-push-container:
needs: [ "publish-artifacts" ]
runs-on: dind
steps:
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Log into Docker Package Registry
uses: docker/login-action@v3
with:
registry: forgejo.neshweb.net
username: ${{ secrets.FORGEJO_USERNAME }}
password: ${{ secrets.FORGEJO_TOKEN }}
- name: Build and push to Docker Package Registry
uses: docker/build-push-action@v5
with:
build-args: |
PACKAGE_VERSION=${{ github.ref_name }}
push: true
tags: forgejo.neshweb.net/firq/dockge-cli:${{ github.ref_name }}
release:
needs: [ build-and-push-container, publish-artifacts ]
if: success()
runs-on: docker
steps:
- name: Release New Version
uses: actions/forgejo-release@v1
with:
direction: upload
url: https://forgejo.neshweb.net
release-dir: release
token: ${{ secrets.FORGEJO_TOKEN }}
tag: ${{ github.ref_name }}

View file

@ -1,34 +0,0 @@
on:
push:
branches: "**"
jobs:
pylint:
runs-on: docker
container: nikolaik/python-nodejs:python3.11-nodejs21
steps:
- name: Checkout source code
uses: https://code.forgejo.org/actions/checkout@v3
- name: Install packages
run: |
pip install -e .[lint] -q --disable-pip-version-check -q
python -m pip list --format=columns --disable-pip-version-check
- name: Run pylint
run: |
pylint --version
pylint **/*.py --exit-zero --rc-file pyproject.toml
mypy:
runs-on: docker
container: nikolaik/python-nodejs:python3.11-nodejs21
steps:
- name: Checkout source code
uses: https://code.forgejo.org/actions/checkout@v3
- name: Install packages
run: |
pip install -e .[typing] -q --disable-pip-version-check -q
python -m pip list --format=columns --disable-pip-version-check
- name: Run mypy
run: |
mypy --version
mypy .

View file

@ -0,0 +1,20 @@
on:
push:
branches: "**"
jobs:
backend-pylint:
runs-on: docker
container: nikolaik/python-nodejs:python3.11-nodejs21
steps:
- name: Checkout source code
uses: https://code.forgejo.org/actions/checkout@v3
- name: Install packages
run: |
pip install -e . -q
python -m pip list --format=columns --disable-pip-version-check
pip install pylint~=2.17.7 --disable-pip-version-check -q
- name: Run pylint
run: |
pylint --version
pylint **/*.py --exit-zero --rc-file pyproject.toml

View file

@ -1,4 +0,0 @@
FROM forgejo.neshweb.net/ci-docker-images/python-neshweb:3.11
ARG PACKAGE_VERSION=0.1.0
RUN pip install dockge-cli==${PACKAGE_VERSION}

View file

@ -1,67 +1,3 @@
# dockge-cli # dockge-cli
A simple CLI application written in Python for communicating with Dockge using websockets A simple CLI application written in Python for communicating with Dockge using websockets
## Background
Dockge (spoken dock-ge or dockage) is a tool to manage docker-compose stacks from a web ui. It is developed by louislam, who also develops UptimeKuma.
Dockge itself doesn't offer any kind of API or programmatic access, as it is just intended for managing stacks via UI.
My current deployment solution for firq.dev and fgo-ta.com is based on Dockge, and I was over it always having to reload the stack whenever I pushed an update. Instead, I wanted to have this as a separate CI step, automatically redeploying a givens stack.
As Dockge is using a websocket-based system under the hood, it was easy to take a look at how communication occurs. In general, communication is achieved by leveraging socket.io for the data. Since Python already offers a solution for socket.io, it is just a matter of emulating the calls the webui sends and receives.
In the end, this is the current result that works pretty well for my understanding. I am still trying to improve upon some issues (login times out, stability, features), but in general this works as a fine solution for automatic stack updating.
## Installation
Install it from the custom package index using
```shell
pip install --extra-index-url https://forgejo.neshweb.net/api/packages/Firq/pypi/simple/ dockge-cli
```
Alternativly, install it using this repository. When installing for development, make sure to install with the additional dependencies
```shell
pip install -e .[lint,typing]
```
## Usage
Call the CLI using `dockge-cli` or `dockge`.
```shell
usage: dockge_cli [-h] [--version] {host,login,logout,list,status,restart,start,stop,down,update,exit,help}
CLI interface for interacting with Dockge
positional arguments:
{host,login,logout,list,status,restart,start,stop,down,update,exit,help}
options:
-h, --help show this help message and exit
--version show program's version number and exit
```
Help for each individual command can be invoked by calling `dockge-cli help <command>`
## The magic behind this
Generally, this makes use of the underlying Websockets API that the Dockge frontend uses to communicate with the server. By analyzing the traffic and looking into the codebase, I was able to reverse most of the packets that are being sent. This allows me to then contruct, send and receive my own packets, making the whole thing work.
There are some things that need to be taken into account for this: For one, dockge uses socket.io for the websocket communication. This meant I had to find the corresponding socket.io version to get the correct version of python-socketio. In addition, I had to find out how the authorization mechanism behind this works.
After finishing up the first prototype, the workings are as follows:
1. A websocket session is established using socket.io - this happens automatically
2. After the session is ready, the `login` command is sent together with a provided username and password
3. Once the CLI is authorized, the selected command is sent
4. The CLI waits for any response values and exits once the command has executed successfully
To provide a smooth experience, both the credentials and the remote host URI are stored on disk. just like the `docker` cli, the credentials are not encrypted, meaning it is advised to either clear the credentials after use OR to use the `--username` and `--password` parameters. This is especially recommended for CI applications.
## Known issues
This CLI does not work when Mullvad is used, as Mullvad actively blocks port forwarding (which python-socketio uses)

1
dockge_cli/.gitignore vendored Normal file
View file

@ -0,0 +1 @@
.temp/*

View file

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

View file

@ -1,50 +1,33 @@
from urllib.parse import urlparse from urllib.parse import urlparse
from getpass import getpass from getpass import getpass
import re
from ...models import Credentials from ...models.parser import Credentials
from ...service import storage from ...service import storage
from ...service.connection import DockgeConnection from ..utils import stack_formatter, status_formatter, generic_formatter, get_credential_parser
from ..utils import stack_formatter, status_formatter, generic_formatter, credential_parser_factory from ...service.communicate import DockgeConnection
class FunctionBindings(): 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():
"""
Creates a connection and logs into Dockge
"""
con = DockgeConnection() con = DockgeConnection()
con.connect_and_login() con.connect_and_login()
return con return con
@staticmethod @staticmethod
def host(extra_args): def host(extra_args):
"""
host command binding
"""
if len(extra_args) > 0: if len(extra_args) > 0:
mat = re.search(r"((\w+\.)?\w+\.\w+(\/.+)?)", extra_args[0], re.IGNORECASE) res = urlparse(extra_args[0])
if mat is None:
raise ValueError("Given host did not match regex")
res = urlparse(f"https://{mat[0]}")
if all([res.scheme, res.netloc]): if all([res.scheme, res.netloc]):
storage.put("host", mat[0]) host = extra_args[0].rstrip("/").replace("https://", "").replace("wss://", "")
storage.put("host", host)
else: else:
raise ValueError(f"Malformed URL {extra_args[0]}") raise ValueError(f"Malformed URL {extra_args[0]}")
print(storage.get("host")) print(storage.get("host"))
@staticmethod @staticmethod
def login(extra_args): def login(extra_args):
"""
login command binding
"""
print(f"WARNING! These credentials will be saved unencrypted in {storage._file.absolute()}")
if len(extra_args) > 0: if len(extra_args) > 0:
credentials = credential_parser_factory().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)
storage.put("password", credentials.password, encoded=True) storage.put("password", credentials.password, encoded=True)
return return
@ -53,85 +36,67 @@ class FunctionBindings():
@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(_):
""" con = ExecutionCommands.__setup()
list command binding
"""
con = FunctionBindings.__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):
""" if extra_args is None:
status command binding raise ValueError
""" con = ExecutionCommands.__setup()
con = FunctionBindings.__setup()
status_formatter(con.list_stack(extra_args[0])) status_formatter(con.list_stack(extra_args[0]))
con.disconnect() con.disconnect()
@staticmethod @staticmethod
def restart(extra_args): def restart(extra_args):
""" if extra_args is None:
restart command binding raise ValueError
""" con = ExecutionCommands.__setup()
con = FunctionBindings.__setup()
generic_formatter(con.restart(extra_args[0])) generic_formatter(con.restart(extra_args[0]))
con.disconnect() con.disconnect()
@staticmethod @staticmethod
def update(extra_args): def update(extra_args):
""" if extra_args is None:
update command binding raise ValueError
""" con = ExecutionCommands.__setup()
con = FunctionBindings.__setup()
generic_formatter(con.update(extra_args[0])) generic_formatter(con.update(extra_args[0]))
con.disconnect() con.disconnect()
@staticmethod @staticmethod
def stop(extra_args): def stop(extra_args):
""" if extra_args is None:
stop command binding raise ValueError
""" con = ExecutionCommands.__setup()
con = FunctionBindings.__setup()
generic_formatter(con.stop(extra_args[0])) generic_formatter(con.stop(extra_args[0]))
con.disconnect() con.disconnect()
@staticmethod @staticmethod
def start(extra_args): def start(extra_args):
""" if extra_args is None:
start command binding raise ValueError
""" con = ExecutionCommands.__setup()
con = FunctionBindings.__setup()
generic_formatter(con.start(extra_args[0])) generic_formatter(con.start(extra_args[0]))
con.disconnect() con.disconnect()
@staticmethod @staticmethod
def down(extra_args): def down(extra_args):
""" if extra_args is None:
down command binding raise ValueError
""" con = ExecutionCommands.__setup()
con = FunctionBindings.__setup()
generic_formatter(con.down(extra_args[0])) generic_formatter(con.down(extra_args[0]))
con.disconnect() con.disconnect()
@staticmethod @staticmethod
def help(): def help():
"""
exit command binding - This should never be invoked
"""
print("WTF") print("WTF")

View file

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

View file

@ -0,0 +1,17 @@
from typing import List, Callable
from pydantic import BaseModel
from .descriptors import command_mappings
class Command(BaseModel):
command: str
description: str
args: int
optional: bool
binding: Callable
commands: dict[str, Command] = {}
descriptors: List[dict[str, object]] = command_mappings
for descriptor in descriptors:
c = Command(**descriptor) # type: ignore
commands.update({ c.command: c })

View file

@ -1,4 +0,0 @@
from ...models import Command
from .mappings import mapping
commands: dict[str, Command] = { c.cmd: c for c in mapping }

View file

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

View file

@ -2,14 +2,10 @@ import argparse
import sys import sys
from .. import __version__ from .. import __version__
from ..models import Arguments from ..models.parser import Arguments
from .commands 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,9 +1,6 @@
from .commands 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
@ -14,10 +11,6 @@ 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")
@ -32,4 +25,4 @@ def run(command, args):
display_help(args) display_help(args)
return return
c.func(args) c.binding(args)

View file

@ -2,10 +2,7 @@ import argparse
from tabulate import tabulate from tabulate import tabulate
from ..models import StackStatus from ..models import StackStatus
def credential_parser_factory(): 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"
@ -15,30 +12,21 @@ def credential_parser_factory():
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")
table, headers = [], ["Stackname", "Status"] table, headers = [], ["Stackname", "Status"]
for key, val in stacks["stackList"].items(): for key, val in stacks["stackList"].items():
table.append([key, StackStatus(val['status']).name.lower()]) table.append([key, StackStatus(val['status']).name])
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,8 +2,5 @@ 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,3 +1 @@
from .codes import StackStatus from .codes import StackStatus
from .command import Command
from .parser import Arguments, Credentials

View file

@ -1,9 +1,7 @@
from enum import Enum from enum import Enum
class StackStatus(Enum): class StackStatus(Enum):
""" # pylint: disable=invalid-name
mapping for plaintext vs statuscode inactive = 1
""" running = 3
INACTIVE = 1 exited = 4
RUNNING = 3
EXITED = 4

View file

@ -1,12 +0,0 @@
from typing import Callable
from pydantic import BaseModel
class Command(BaseModel):
"""
Basic command structure for the CLI to automatically generate valid commands
"""
cmd: str
func: Callable
args: int
optional: bool
description: str

View file

@ -2,15 +2,9 @@ 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

@ -1 +0,0 @@
.storage/*

View file

@ -5,14 +5,8 @@ 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
@ -32,7 +26,7 @@ class DockgeConnection:
def _init_events(self): def _init_events(self):
@self._sio.event @self._sio.event
def connect(): def connect():
self.login() self.connect()
print("Connected!") print("Connected!")
@self._sio.event @self._sio.event
@ -63,34 +57,20 @@ class DockgeConnection:
success = True success = True
else: else:
print("Issue with login procedure") print("Issue with login procedure")
print(data)
return success return success
# Functions # Functions
def connect_and_login(self): def connect_and_login(self):
""" self._sio.connect(f"https://{self._host}/socket.io/", transports=['websocket'])
Connect to the websocket self.connect()
"""
# Dockge uses Socket.io for the websockets, so this URI and params are always the same
self._sio.connect(f"https://{self._host}/socket.io/")
self.login()
def login(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
data = None data = None
retry, count = True, 0 retry, count = True, 0
if not storage.exists("username"):
raise ValueError("Missing username")
if not storage.exists("password"):
raise ValueError("Missing password")
while retry and count < 5: while retry and count < 5:
try: try:
data = self._sio.call( data = self._sio.call(
@ -98,13 +78,12 @@ class DockgeConnection:
{ {
"username": storage.get("username", encoded=True), "username": storage.get("username", encoded=True),
"password": storage.get("password", encoded=True), "password": storage.get("password", encoded=True),
"token": "" "token":""
}, },
timeout=10 timeout=5
) )
retry = False retry = False
except socketio.exceptions.TimeoutError: except socketio.exceptions.TimeoutError:
print("Reached timeout for login, retrying ...")
retry = True retry = True
count += 1 count += 1
@ -113,9 +92,6 @@ 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)
@ -124,50 +100,29 @@ class DockgeConnection:
return retval return retval
def list_stack(self, name: str): def list_stack(self, name: str):
""" ret = self._sio.call("agent", ("", "serviceStatusList", name), timeout=5)
Lists status for a stack
"""
ret = self._sio.call("agent", ("", "serviceStatusList", name), timeout=10)
return ret return ret
def restart(self, name): def restart(self, name):
""" ret = self._sio.call("agent", ("", "restartStack", name), timeout=10)
Restarts a given stack
"""
ret = self._sio.call("agent", ("", "restartStack", name), timeout=30)
return ret return ret
def update(self, name): def update(self, name):
""" ret = self._sio.call("agent", ("", "updateStack", name), timeout=10)
Updates a given stack
"""
ret = self._sio.call("agent", ("", "updateStack", name), timeout=30)
return ret return ret
def stop(self, name): def stop(self, name):
""" ret = self._sio.call("agent", ("", "stopStack", name), timeout=10)
Stops a given stack
"""
ret = self._sio.call("agent", ("", "stopStack", name), timeout=30)
return ret return ret
def start(self, name): def start(self, name):
""" ret = self._sio.call("agent", ("", "startStack", name), timeout=10)
Starts a given stack
"""
ret = self._sio.call("agent", ("", "startStack", name), timeout=30)
return ret return ret
def down(self, name): def down(self, name):
""" ret = self._sio.call("agent", ("", "downStack", name), timeout=10)
Stops and downs a given stack
"""
ret = self._sio.call("agent", ("", "downStack", name), timeout=30)
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

@ -3,82 +3,42 @@ import pathlib
import base64 import base64
import yaml import yaml
_storagepath = pathlib.Path(__file__).parent / ".storage" _storagepath = pathlib.Path(__file__).parents[1] / ".temp"
_storagepath.mkdir(exist_ok=True, parents=True)
_file = _storagepath / "storage.yaml" _file = _storagepath / "storage.yaml"
def create_file_when_missing(): _storagepath.mkdir(exist_ok=True, parents=True)
"""
Checks if storage file does exist, creates it when necessary
"""
if _file.exists():
return
with open(_file, 'a', encoding="utf-8"):
os.utime(_file, None)
def exists(key: str) -> bool: def fileexists():
"""
Checks if a given key exists in the storage file
"""
if not _file.exists(): if not _file.exists():
return False with open(_file, 'a', encoding="utf-8"):
os.utime(_file, None)
with open(_file, "r", encoding="utf-8") as file:
content: dict[str, str] = yaml.load(file, Loader=yaml.SafeLoader)
return key in content
def put(key: str, value: str, encoded=False): def put(key: str, value: str, encoded=False):
""" fileexists()
Puts a given value with a given key into the storage file with open(_file, "r+", encoding="utf-8") as file:
Encodes the data as base64 when encoded is set to true
"""
if not _file.exists():
create_file_when_missing()
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: str(base64.b64encode(value.encode()), "utf-8") if encoded else value }) content.update({ key: str(base64.b64encode(value.encode())) if encoded else value })
with open(_file, "w+", encoding="utf-8") 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()
Removed a given key from the storage file
"""
if not _file.exists():
create_file_when_missing()
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 {}
content.pop(key, None) content.pop(key, None)
with open(_file, "w+", encoding="utf-8") 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):
"""
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
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) 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:
return None return None
return base64.b64decode(value.encode()).decode() if encoded else value return base64.b64decode(value).decode() if encoded else value
def clear(): def clear():
"""
Deletes the storage file
"""
if not _file.exists():
return
_file.unlink() _file.unlink()

View file

@ -1,10 +1,9 @@
[project] [project]
name = "dockge_cli" name = "dockge_cli"
version = "0.1.2" version = "0.0.1-c.1"
dependencies = [ dependencies = [
"pyyaml~=6.0.1", "pyyaml~=6.0.1",
"pydantic~=2.8.0", "pydantic~=2.8.0",
"requests~=2.32.3",
"python-socketio~=5.11.3", "python-socketio~=5.11.3",
"websocket-client~=1.8.0", "websocket-client~=1.8.0",
"tabulate ~=0.9.0", "tabulate ~=0.9.0",
@ -12,7 +11,7 @@ dependencies = [
requires-python = ">= 3.10" requires-python = ">= 3.10"
authors = [{name = "Firq", email = "firelp42@gmail.com"}] authors = [{name = "Firq", email = "firelp42@gmail.com"}]
maintainers = [{name = "Firq", email = "firelp42@gmail.com"}] maintainers = [{name = "Firq", email = "firelp42@gmail.com"}]
description = "CLI for interacting with dockge" description = "CLi for interacting with dockge"
classifiers = [ classifiers = [
"Development Status :: 3 - Alpha", "Development Status :: 3 - Alpha",
"Programming Language :: Python :: 3", "Programming Language :: Python :: 3",
@ -32,30 +31,8 @@ include = ["dockge_cli*"]
[tool.setuptools.package-data] [tool.setuptools.package-data]
"*" = ["py.typed"] "*" = ["py.typed"]
[project.optional-dependencies]
lint = [
"pylint~=3.2.5",
]
typing = [
"mypy~=1.10.1",
"types-PyYAML~=6.0.12.20240311",
"types-tabulate~=0.9.0.20240106",
]
[tool.pylint."MAIN"] [tool.pylint."MAIN"]
disable = [ disable = [ "line-too-long", "missing-module-docstring", "missing-function-docstring", "missing-class-docstring" ]
"line-too-long",
"missing-module-docstring",
]
[tool.mypy]
python_version = "3.11"
warn_return_any = true
warn_unused_configs = true
[[tool.mypy.overrides]]
module = 'socketio.*'
ignore_missing_imports = true
[build-system] [build-system]
requires = ["setuptools >= 61.0"] requires = ["setuptools >= 61.0"]