first commit

config/settings.py

@@ -0,0 +1,59 @@
+import os
+from xdg.BaseDirectory import xdg_config_home
+from configparser import ConfigParser
+
+
+class Config:
+    def __init__(self):
+        self.config = ConfigParser()
+        self.config_dir = os.path.join(xdg_config_home, "natalie")
+
+    def check_dir_exists(self):
+        if not os.path.exists(self.config_dir):
+            os.makedirs(self.config_dir)
+
+    def check_file_exists(self, filename):
+        config_file = os.path.join(self.config_dir, filename)
+        return os.path.exists(config_file)
+
+
+class PortainerConfig(Config):
+    def __init__(self):
+        super().__init__()
+        self.config_file = os.path.join(self.config_dir, "portainer.ini")
+        self.config.read(self.config_file)
+
+    def get_portainer_instances(self):
+        return self.config.sections()
+
+    def set_new_portainer(self, name, base_url, token):
+        self.config[name] = {"url": base_url, "token": token}
+        with open(self.config_file, 'w') as configfile:
+            self.config.write(configfile)
+
+    def get_portainer_info(self, name):
+        base_url = self.config[name]['url']
+        token = self.config[name]['token']
+        if base_url and token:
+            return base_url, token
+        else:
+            return None
+
+
+class DockerConfig(Config):
+    def __init__(self):
+        super().__init__()
+        self.config_file = os.path.join(self.config_dir, 'docker.ini')
+        self.config.read(self.config_file)
+
+    def get_docker_hosts_list(self):
+        docker_hosts_list = self.config['hosts'].keys()
+        return list(docker_hosts_list)
+
+    def get_docker_host(self, hostname):
+        return self.config['hosts'][hostname]
+
+    def set_new_docker_host(self, hostname, url, port):
+        self.config['hosts'][hostname] = f"{url}:{port}"
+        with open(self.config_file, 'w') as configfile:
+            self.config.write(configfile)

docker_api/api/__init__.py

@@ -0,0 +1,9 @@
+from .containers import Containers
+from .images import Images
+
+
+class DockerManager:
+    def __init__(self, docker_host):
+        self.docker_host = docker_host
+        self.containers = Containers(docker_host)
+        self.images = Images(docker_host)

docker_api/api/base.py

@@ -0,0 +1,6 @@
+import docker
+
+
+class Docker:
+    def __init__(self, endpoint):
+        self.client = docker.DockerClient(endpoint)

docker_api/api/containers.py

@@ -0,0 +1,7 @@
+from .base import Docker
+
+
+class Containers(Docker):
+    def list_containers(self):
+        containers = self.client.containers.list()
+        return containers

docker_api/api/images.py

@@ -0,0 +1,11 @@
+from .base import Docker
+
+
+class Images(Docker):
+    def list_images(self):
+        images = self.client.images.list()
+        return images
+
+    def pull_image(self, repository):
+        image = self.client.images.pull(repository)
+        return image

docker_api/docker_manager.py

@@ -0,0 +1,43 @@
+from .api import DockerManager
+from config.settings import DockerConfig
+from tui.terminal_menu import terminal_menu
+
+
+def docker_manager_menu():
+    config = DockerConfig()
+    docker_hosts = config.get_docker_hosts_list()
+    docker_hosts.append('Add Docker host')
+    docker_hosts.append('Back to main menu')
+    selected_docker_host = terminal_menu(docker_hosts, "Docker hosts")
+    while selected_docker_host != "Back to main menu":
+        if selected_docker_host == "Add Docker host":
+            host_name = input("Enter Docker host name: ")
+            url = input("Enter Docker host URL: ")
+            port = input("Enter Docker host port: ")
+            config.set_new_docker_host(host_name, url, port)
+        else:
+            docker_host = config.get_docker_host(selected_docker_host)
+            docker_manager = DockerManager(docker_host)
+            docker_menu = [
+                "List containers",
+                "List images",
+                "Update images",
+                "Back to Docker hosts"
+            ]
+            selected_docker_menu_option = terminal_menu(docker_menu, docker_host)
+            while selected_docker_menu_option != "Back to Docker hosts":
+                if selected_docker_menu_option == "List containers":
+                    containers_list = docker_manager.containers.list_containers()
+                    for c in containers_list:
+                        image = c.attrs['Config']['Image']
+                        print(f"Container ID: {c.id[:12]} - Image: {image}")
+                    update_confirm = input("Do you want to update images (y/n): ")
+                    if update_confirm.lower() == 'y':
+                        for c in containers_list:
+                            print(f"Updating container: {c.id[:12]}")
+                            image = docker_manager.images.pull_image(c.attrs['Config']['Image'])
+                            print(f"Pulled image {image.tags[0]}")
+                        print(f"Finished updating {len(containers_list)} images")
+
+                selected_docker_menu_option = terminal_menu(docker_menu, docker_host)
+        selected_docker_host = terminal_menu(docker_hosts, "Docker hosts")

gui/portainer_ui.py

@@ -0,0 +1,67 @@
+import sys
+from PyQt6.QtWidgets import (QApplication, QMainWindow, QWidget,
+                             QVBoxLayout, QHBoxLayout,
+                             QLabel, QLineEdit, QListWidget,
+                             QTableWidget, QHeaderView, QPushButton, QComboBox)
+
+
+class PortainerUi:
+    def __init__(self):
+        self.main_window = QMainWindow()
+
+        self.main_window.show()
+        self.main_window.setWindowTitle("Natalie")
+
+        self.search_box = QLineEdit()
+        self.combo_box = QComboBox()
+
+        search_layout = QHBoxLayout()
+        search_layout.addWidget(self.search_box)
+        search_layout.addWidget(self.combo_box)
+
+        self.stacks_list = QListWidget()
+
+        self.stack_title = QLabel()
+        self.containers_table = QTableWidget()
+        self.containers_table.setColumnCount(3)
+        self.containers_table.setHorizontalHeaderLabels(["ID", "Name", "Status"])
+        self.containers_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
+        self.containers_table.horizontalHeader().setStretchLastSection(True)
+
+        self.start = QPushButton("Start")
+        self.stop = QPushButton("Stop")
+        self.update = QPushButton("Update")
+
+        buttons_layout = QHBoxLayout()
+        buttons_layout.addWidget(self.start)
+        buttons_layout.addWidget(self.stop)
+        buttons_layout.addWidget(self.update)
+
+        info_layout = QVBoxLayout()
+        info_layout.addWidget(self.stack_title)
+        info_layout.addLayout(buttons_layout)
+        info_layout.addWidget(self.containers_table)
+
+        main_layout = QHBoxLayout()
+        main_layout.addWidget(self.stacks_list)
+        main_layout.addLayout(info_layout)
+        main_layout.setStretch(0, 1)
+        main_layout.setStretch(1, 2)
+
+        layout = QVBoxLayout()
+        layout.addLayout(search_layout)
+        layout.addLayout(main_layout)
+
+        widget = QWidget()
+        widget.setLayout(layout)
+        self.main_window.setCentralWidget(widget)
+
+    def show(self):
+        self.main_window.show()
+
+
+if __name__ == '__main__':
+    app = QApplication(sys.argv)
+    win = PortainerUi()
+    win.show()
+    sys.exit(app.exec())

gui/workers/load_endpoints_worker.py

@@ -0,0 +1,17 @@
+from PyQt6.QtCore import QThread, pyqtSignal
+
+
+class LoadEndpointsWorker(QThread):
+    endpoints_retrieved = pyqtSignal(list)
+    error_occurred = pyqtSignal(str)
+
+    def __init__(self, portainer):
+        super().__init__()
+        self.portainer = portainer
+
+    def run(self):
+        try:
+            endpoints = self.portainer.endpoints.list_endpoints()
+            self.endpoints_retrieved.emit(endpoints)
+        except Exception as e:
+            self.error_occurred.emit(str(e))

gui/workers/load_stacks_worker.py

@@ -0,0 +1,18 @@
+from PyQt6.QtCore import QThread, pyqtSignal
+from portainer_api import PortainerService
+
+
+class LoadStacksWorker(QThread):
+    stacks_retrieved = pyqtSignal(list)
+    error_occurred = pyqtSignal(str)
+
+    def __init__(self, portainer):
+        super().__init__()
+        self.portainer = portainer
+
+    def run(self):
+        try:
+            stacks = self.portainer.stacks.list_stacks()
+            self.stacks_retrieved.emit(stacks)
+        except Exception as e:
+            self.error_occurred.emit(str(e))

natalie.py

@@ -0,0 +1,51 @@
+import sys
+from PyQt6.QtWidgets import (QApplication, QListWidgetItem)
+from PyQt6.QtCore import Qt
+from portainer_api import PortainerService
+from gui.portainer_ui import PortainerUi
+from config.settings import PortainerConfig
+
+from gui.workers.load_stacks_worker import LoadStacksWorker
+from gui.workers.load_endpoints_worker import LoadEndpointsWorker
+
+
+class Natalie(PortainerUi):
+    def __init__(self, portainer_url, portainer_token):
+        super().__init__()
+        self.portainer = PortainerService(portainer_url, portainer_token)
+        self.endpoints_worker = LoadEndpointsWorker(self.portainer)
+        self.endpoints_worker.endpoints_retrieved.connect(self.load_endpoints)
+        self.endpoints_worker.error_occurred.connect(self.error_occurred)
+        self.endpoints_worker.start()
+        self.stacks_worker = LoadStacksWorker(self.portainer)
+        self.stacks_worker.stacks_retrieved.connect(self.load_stacks)
+        self.stacks_worker.error_occurred.connect(self.error_occurred)
+        self.stacks_worker.start()
+
+    def load_endpoints(self, endpoints):
+        self.endpoints_list = endpoints
+        self.combo_box.clear()
+        self.combo_box.addItem("All")
+        for endpoint in endpoints:
+            self.combo_box.addItem(endpoint['name'])
+
+    def load_stacks(self, stacks):
+        current_group = self.combo_box.currentText()
+        self.stacks_list.clear()
+        for stack in stacks:
+            item = QListWidgetItem(stack['name'])
+            item.setData(Qt.ItemDataRole.UserRole, stack)
+            self.stacks_list.addItem(item)
+
+    def error_occurred(self, error_message):
+        print(f"Error occurred: {error_message}")
+
+
+if __name__ == '__main__':
+    portainer_config = PortainerConfig()
+    instances = portainer_config.get_portainer_instances()
+    base_url, token = portainer_config.get_portainer_info(instances[0])
+    app = QApplication(sys.argv)
+    win = Natalie(base_url, token)
+    win.show()
+    sys.exit(app.exec())

natalie_cli.py

@@ -0,0 +1,16 @@
+import argparse
+from portainer_api import PortainerService
+from config.settings import PortainerConfig
+from docker_api.docker_manager import docker_manager_menu
+
+if __name__ == '__main__':
+    portainer_config = PortainerConfig()
+    instances = portainer_config.get_portainer_instances()
+    base_url, token = portainer_config.get_portainer_info(instances[0])
+    print(base_url, token)
+    portainer = PortainerService(base_url, token)
+    stacks = portainer.stacks.list_stacks()
+    for stack in stacks:
+        print(stack)
+
+    docker_manager_menu()

portainer_api/__init__.py

@@ -0,0 +1,8 @@
+from .stacks import Stacks
+
+
+class PortainerService:
+    def __init__(self, base_url, token):
+        self.base_url = base_url
+        self.token = token
+        self.stacks = Stacks(base_url, token)

portainer_api/base.py

@@ -0,0 +1,6 @@
+from portainer import PortainerApi
+
+
+class Portainer:
+    def __init__(self, base_url, token):
+        self.portainer = PortainerApi(base_url, token)

portainer_api/endpoints.py

@@ -0,0 +1,18 @@
+from .base import Portainer
+
+
+class Endpoints(Portainer):
+    def list_endpoints(self):
+        endpoints = self.portainer.endpoints.list_endpoints()
+        endpoints_list = []
+        for endpoint in endpoints:
+            endpoint_data = {
+                "id": endpoint['Id'],
+                "name": endpoint['Name'],
+                "url": endpoint['URL'],
+                "status": endpoint['Status'],
+            }
+            endpoints_list.append(endpoint_data)
+
+        return endpoints_list
+    

portainer_api/stacks.py

@@ -0,0 +1,44 @@
+from .base import Portainer
+
+
+class Stacks(Portainer):
+    def list_stacks(self):
+        stacks = self.portainer.stacks.list_stacks()
+        stacks_list = []
+        for stack in stacks:
+            stack_data = {
+                "id": stack['Id'],
+                "name": stack['Name'],
+                "type": stack['Type'],
+                "endpoint_id": stack['EndpointId'],
+                "resource_id": stack['ResourceControl']['ResourceId'],
+                "status": stack['Status'],
+                "creation_date": stack['CreationDate'],
+                "created_by": stack['CreatedBy'],
+                "update_date": stack['UpdateDate'],
+                "updated_by": stack['UpdatedBy'],
+                "webhook": stack['Webhook']
+            }
+            stacks_list.append(stack_data)
+        return stacks_list
+
+    def get_stack_by_id(self, stack_id):
+        stack = self.portainer.stacks.get_stack_by_id(stack_id)
+        stack_data = {
+            "id": stack['Id'],
+            "name": stack['Name'],
+            "type": stack['Type'],
+            "endpoint_id": stack['EndpointId'],
+            "resource_id": stack['ResourceControl']['ResourceId'],
+            "status": stack['Status'],
+            "creation_date": stack['CreationDate'],
+            "created_by": stack['CreatedBy'],
+            "update_date": stack['UpdateDate'],
+            "updated_by": stack['UpdatedBy'],
+            "webhook": stack['Webhook']
+        }
+        return stack_data
+
+    def update_stack(self, webhook):
+        response = self.portainer.stacks.update_stack(webhook)
+        return response

tui/terminal_menu.py

@@ -0,0 +1,21 @@
+from simple_term_menu import TerminalMenu
+
+
+def terminal_menu(menu_options, menu_title, clear_screen=False):
+    if isinstance(menu_options, list):
+        menu = TerminalMenu(menu_options, clear_screen=clear_screen, title=menu_title)
+        entry_index = menu.show()
+        if entry_index is not None:
+            return menu_options[entry_index]
+
+    elif isinstance(menu_options, dict):
+        menu = TerminalMenu(list(menu_options.keys()), clear_screen=True, title=menu_title)
+        entry_index = menu.show()
+        if entry_index is not None:
+            selected_key = list(menu_options.keys())[entry_index]
+            return selected_key, menu_options[selected_key]
+        return None
+
+    else:
+        raise ValueError("Input must be list or dict.")
+