jaydee/portainer
1
0
Fork 0
mirror of https://gitlab.sectorq.eu/jaydee/portainer.git synced 2025-12-14 10:44:52 +01:00
Code Issues Packages Projects Releases Wiki Activity

build

Browse Source
This commit is contained in:
Ladislav Dusa
2025-12-02 20:00:22 +01:00
parent dec2f81b05
commit e052fd7038
2 changed files with 127 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
port.py
View File

@@ -2,15 +2,15 @@ import os
import requests import requests
import json import json
import uuid import uuid
import argparse
import shutil import shutil
import time import time
import logging import logging
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
from tabulate import tabulate
from git import Repo # pip install gitpython
import base64
class Portainer: class Portainer:
""" """
Simple wrapper around the module-level Portainer helper functions. Simple wrapper around the module-level Portainer helper functions.
@@ -28,22 +28,83 @@ class Portainer:
self.stack_ids = [] self.stack_ids = []
self.endpoint_name = None self.endpoint_name = None
self.endpoint_id = None self.endpoint_id = None
#self.git_url = "https://gitlab.sectorq.eu/home/docker-compose.git" # self.git_url = "https://gitlab.sectorq.eu/home/docker-compose.git"
self.git_url = "git@gitlab.sectorq.eu:home/docker-compose.git" self.git_url = "git@gitlab.sectorq.eu:home/docker-compose.git"
self.repo_dir = "/tmp/docker-compose" self.repo_dir = "/tmp/docker-compose"
self.basic_stacks = ["pihole","nginx", "mosquitto", "webhub", "authentik","bitwarden","mailu3","home-assistant","homepage"] self.basic_stacks = [
self.nas_stacks = self.basic_stacks + ["gitlab", "bookstack","dockermon","gitea","grafana","immich","jupyter","kestra","mealie"] "pihole",
self.m_server_stacks = self.basic_stacks + ['immich', 'zabbix-server', 'gitea', 'unifibrowser', 'mediacenter', 'watchtower', 'wazuh', 'octoprint', 'motioneye', 'kestra', 'bookstack', 'wud', 'uptime-kuma', 'registry', 'regsync', 'dockermon', 'grafana', 'nextcloud', 'semaphore', 'node-red', 'test', 'jupyter', 'paperless', 'mealie', 'n8n', 'ollama', 'rancher'] "nginx",
self.rpi5_stacks = self.basic_stacks + ["gitlab","bookstack","gitea"] "mosquitto",
self.rack_stacks = self.basic_stacks + ["gitlab", "bookstack","dockermon","gitea","grafana","immich","jupyter","kestra","mealie"] "webhub",
"authentik",
"bitwarden",
"mailu3",
"home-assistant",
"homepage"
]
self.nas_stacks = self.basic_stacks + [
"gitlab",
"bookstack",
"dockermon",
"gitea",
"grafana",
"immich",
"jupyter",
"kestra",
"mealie"
]
self.m_server_stacks = self.basic_stacks + [
'immich',
'zabbix-server',
'gitea',
'unifibrowser',
'mediacenter',
'watchtower',
'wazuh',
'octoprint',
'motioneye',
'kestra',
'bookstack',
'wud',
'uptime-kuma',
'registry',
'regsync',
'dockermon',
'grafana',
'nextcloud',
'semaphore',
'node-red',
'test',
'jupyter',
'paperless',
'mealie',
'n8n',
'ollama',
'rancher'
]
self.rpi5_stacks = self.basic_stacks + [
"gitlab",
"bookstack",
"gitea"
]
self.rack_stacks = self.basic_stacks + [
"gitlab",
"bookstack",
"dockermon",
"gitea",
"grafana",
"immich",
"jupyter",
"kestra",
"mealie"
]
self.log_mode = False self.log_mode = False
self.hw_mode = False self.hw_mode = False
self.all_data = {"containers":{},"stacks":{},"endpoints":{}} self.all_data = {"containers": {}, "stacks": {}, "endpoints": {}}
self.get_endpoints() self.get_endpoints()
self.get_stacks() self.get_stacks()
self.get_containers() self.get_containers()
def is_number(self, s): def is_number(self, s):
"""Check if the input string is a number.""" """Check if the input string is a number."""
try: try:
@@ -62,11 +123,11 @@ class Portainer:
def api_post(self, path, json="", timeout=120): def api_post(self, path, json="", timeout=120):
url = f"{self.base_url.rstrip('/')}{path}" url = f"{self.base_url.rstrip('/')}{path}"
headers = {"X-API-Key": f"{self.token}"} headers = {"X-API-Key": f"{self.token}"}
#print(url) # print(url)
#print(json) # print(json)
resp = requests.post(url, headers=headers, json=json, timeout=timeout) resp = requests.post(url, headers=headers, json=json, timeout=timeout)
return resp.text return resp.text
def api_post_file(self, path, endpoint_id, name, envs, file, timeout=120): def api_post_file(self, path, endpoint_id, name, envs, file, timeout=120):
#input("API POST2 called. Press Enter to continue.") #input("API POST2 called. Press Enter to continue.")
"""Example authenticated GET request to Portainer API.""" """Example authenticated GET request to Portainer API."""
@@ -361,18 +422,15 @@ class Portainer:
self.endpoint_id = self.get_endpoint_id(endpoint) self.endpoint_id = self.get_endpoint_id(endpoint)
if os.path.exists(self.repo_dir): if os.path.exists(self.repo_dir):
shutil.rmtree(self.repo_dir) shutil.rmtree(self.repo_dir)
print(f"Folder '{self.repo_dir}' has been removed.")
else: else:
print(f"Folder '{self.repo_dir}' does not exist.") print(f"Folder '{self.repo_dir}' does not exist.")
Repo.clone_from(self.git_url, self.repo_dir) Repo.clone_from(self.git_url, self.repo_dir)
if mode == "git": if mode == "git":
print("Creating new stack from git repo...")
path = f"/stacks/create/{p}/repository" path = f"/stacks/create/{p}/repository"
if self.endpoint_id is not None: if self.endpoint_id is not None:
path += f"?endpointId={self.endpoint_id}" path += f"?endpointId={self.endpoint_id}"
print(path)
if stack == "all": if stack == "all":
if self.endpoint_name == "rack": if self.endpoint_name == "rack":

83
portainer.py
View File

@@ -24,8 +24,8 @@ defaults = {
parser = argparse.ArgumentParser(description="Portainer helper - use env vars or pass credentials.") parser = argparse.ArgumentParser(description="Portainer helper - use env vars or pass credentials.")
parser.add_argument("--base", "-b", default=os.getenv("PORTAINER_URL", "https://portainer.example.com"), parser.add_argument("--base", "-b", default=os.getenv("PORTAINER_URL", \
help="Base URL for Portainer (ENV: PORTAINER_URL)") "https://portainer.example.com"),help="Base URL for Portainer (ENV: PORTAINER_URL)")
parser.add_argument("--site", "-t", type=str, default=None, help="Site") parser.add_argument("--site", "-t", type=str, default=None, help="Site")
parser.add_argument("--endpoint-id", "-e", type=str, default=None, help="Endpoint ID to limit stack operations") parser.add_argument("--endpoint-id", "-e", type=str, default=None, help="Endpoint ID to limit stack operations")
parser.add_argument("--refresh-environment", "-R", action="store_true", help="List endpoints") parser.add_argument("--refresh-environment", "-R", action="store_true", help="List endpoints")
@@ -114,7 +114,24 @@ def get_portainer_token(base_url, username=None, password=None, timeout=10):
if not token: if not token:
raise ValueError(f"No token found in response: {data}") raise ValueError(f"No token found in response: {data}")
return token return token
def prompt_missing_args(args, defaults, fields):
"""
fields = [("arg_name", "Prompt text")]
"""
for field, text in fields:
value = getattr(args, field)
default = defaults.get(field)
if value is None:
if default is not None:
prompt = f"{text} (default={default}) : "
value = input(prompt) or default
else:
value = input(f"{text}: ")
setattr(args, field, value)
return args
if __name__ == "__main__": if __name__ == "__main__":
# Example usage: set PORTAINER_USER and PORTAINER_PASS in env, or pass literals below. # Example usage: set PORTAINER_USER and PORTAINER_PASS in env, or pass literals below.
#token = get_portainer_token(base,"admin","l4c1j4yd33Du5lo") # or get_portainer_token(base, "admin", "secret") #token = get_portainer_token(base,"admin","l4c1j4yd33Du5lo") # or get_portainer_token(base, "admin", "secret")
@@ -149,24 +166,24 @@ if __name__ == "__main__":
sys.exit() sys.exit()
if args.action == "delete_stack": if args.action == "delete_stack":
if args.endpoint_id == None: args = prompt_missing_args(args, defaults, [
args.endpoint_id = input("Endpoint ID is required for deleting stacks : ") ("site", "Site"),
if args.stack == None: ("endpoint_id", "Endpoint ID"),
args.stack = input("Stack name or ID is required for deleting stacks : ") ("stack", "Stack name or ID")
])
por.delete_stack(args.endpoint_id, args.stack,) por.delete_stack(args.endpoint_id, args.stack,)
sys.exit() sys.exit()
if args.action == "create_stack": if args.action == "create_stack":
if args.endpoint_id == None: args = prompt_missing_args(args, defaults, [
args.endpoint_id = input(f"Endpoint ID (default={defaults["endpoint_id"]}) : ") or defaults["endpoint_id"] ("site", "Site"),
if args.stack == None: ("endpoint_id", "Endpoint ID"),
args.stack = input(f"Stack name or ID : ") ("stack", "Stack name or ID"),
if args.stack_mode == None: ("stack_mode", "Stack mode (swarm or compose)"),
args.stack_mode = input(f"Stack mode (swarm or compose) (default={defaults["stack_mode"]}) : ") or defaults["stack_mode"] ("deploy_mode", "Deploy mode (git or upload)")
if args.deploy_mode == None: ])
args.deploy_mode = input(f"Deploy mode (git or upload) (default={defaults["deploy_mode"]}) : ") or defaults["deploy_mode"]
if args.site == None:
args.site = input(f"Site (default={defaults["site"]}) : ") or defaults["site"]
por.create_stack(args.endpoint_id,args.stack, args.deploy_mode, args.autostart, args.stack_mode) por.create_stack(args.endpoint_id,args.stack, args.deploy_mode, args.autostart, args.stack_mode)
sys.exit() sys.exit()
@@ -183,7 +200,7 @@ if __name__ == "__main__":
args.endpoint_id = input("Endpoint ID is required for starting stacks : ") args.endpoint_id = input("Endpoint ID is required for starting stacks : ")
if args.stack == None: if args.stack == None:
args.stack = input("Stack name or ID is required for starting stacks : ") args.stack = input("Stack name or ID is required for starting stacks : ")
por.start_stack(args.stack,args.endpoint_id) por.start_stack(args.stack, args.endpoint_id)
sys.exit() sys.exit()
if args.action == "list_stacks": if args.action == "list_stacks":
@@ -202,17 +219,17 @@ if __name__ == "__main__":
por.update_stack(args.endpoint_id,args.stack,autostart) por.update_stack(args.endpoint_id,args.stack,autostart)
sys.exit() sys.exit()
if args.action == "print_all_data": if args.action == "print_all_data":
print(json.dumps(por.all_data,indent=2)) print(json.dumps(por.all_data, indent=2))
sys.exit() sys.exit()
if args.action == "update_status": if args.action == "update_status":
por.update_status(args.endpoint_id,args.stack) por.update_status(args.endpoint_id, args.stack)
sys.exit() sys.exit()
if args.action == "list_endpoints": if args.action == "list_endpoints":
eps = por.get_endpoints() eps = por.get_endpoints()
data = [] data = []
for i in eps["by_id"]: for i in eps["by_id"]:
data.append([i,eps["by_id"][i]]) data.append([i, eps["by_id"][i]])
headers = ["EndpointId", "Name"] headers = ["EndpointId", "Name"]
print(tabulate(data, headers=headers, tablefmt="github")) print(tabulate(data, headers=headers, tablefmt="github"))
@@ -226,37 +243,35 @@ if __name__ == "__main__":
cont = [] cont = []
for c in por.all_data["containers"][args.endpoint_id]: for c in por.all_data["containers"][args.endpoint_id]:
if args.stack == c or args.stack == "all": if args.stack == c or args.stack == "all":
cont+=por.all_data["containers"][args.endpoint_id][c] cont += por.all_data["containers"][args.endpoint_id][c]
por.stop_containers(args.endpoint_id,cont) por.stop_containers(args.endpoint_id, cont)
sys.exit() sys.exit()
if args.action == "start_containers": if args.action == "start_containers":
print("Starting containers") print("Starting containers")
cont = [] cont = []
#input(json.dumps(por.all_data,indent=2)) # input(json.dumps(por.all_data, indent=2))
for c in por.all_data["containers"][args.endpoint_id]: for c in por.all_data["containers"][args.endpoint_id]:
if args.stack == c or args.stack == "all": if args.stack == c or args.stack == "all":
cont+=por.all_data["containers"][args.endpoint_id][c] cont += por.all_data["containers"][args.endpoint_id][c]
por.start_containers(args.endpoint_id,cont) por.start_containers(args.endpoint_id, cont)
sys.exit() sys.exit()
if args.action == "start_containers": if args.action == "start_containers":
print("Starting containers") print("Starting containers")
cont = [] cont = []
#input(json.dumps(por.all_data,indent=2)) # input(json.dumps(por.all_data,indent=2))
for c in por.all_data["containers"][args.endpoint_id]: for c in por.all_data["containers"][args.endpoint_id]:
if args.stack == c or args.stack == "all": if args.stack == c or args.stack == "all":
cont+=por.all_data["containers"][args.endpoint_id][c] cont += por.all_data["containers"][args.endpoint_id][c]
por.start_containers(args.endpoint_id,cont) por.start_containers(args.endpoint_id, cont)
sys.exit() sys.exit()
if args.action == "refresh_environment": if args.action == "refresh_environment":
cont = por.refresh() cont = por.refresh()
sys.exit() sys.exit()
if args.action == "refresh_status": if args.action == "refresh_status":
if args.stack== "all": if args.stack == "all":
print("Stopping all stacks...") print("Stopping all stacks...")
stcks = por.get_stacks(base, token, endpoint_id=args.endpoint_id) stcks = por.get_stacks(base, token, endpoint_id=args.endpoint_id)
# stcks = get_stack(base, sta, token, endpoint_id=install_endpoint_id)
else: else:
por.refresh_status(base, args.stack_id, token) por.refresh_status(base, args.stack_id, token)