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

build

Browse Source
This commit is contained in:
Ladislav Dusa
2025-12-12 10:39:06 +01:00
parent 101bfbc9a4
commit 04248ce279
2 changed files with 74 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
port.py
View File

@@ -27,9 +27,10 @@ class Portainer:
to perform API operations. to perform API operations.
""" """
def __init__(self, site, timeout=10): def __init__(self, site, args=None, timeout=120):
self.base_url = None self.base_url = None
self.token = None self.token = None
self.args = args
self.action = None self.action = None
self._debug = False self._debug = False
self.timeout = timeout self.timeout = timeout
@@ -39,7 +40,7 @@ class Portainer:
self.stack_id = None self.stack_id = None
self.stack_ids = [] self.stack_ids = []
self.endpoint_name = None self.endpoint_name = None
self.endpoint_id = None self.endpoint_id = args.endpoint_id
# 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"
@@ -275,9 +276,9 @@ class Portainer:
def get_services(self, endpoint, timeout=30): def get_services(self, endpoint, timeout=30):
'''Get a list of services for a specific stack on an endpoint.''' '''Get a list of services for a specific stack on an endpoint.'''
print(json.dumps(self.all_data,indent=2)) # print(json.dumps(self.all_data,indent=2))
path = f"/endpoints/{endpoint}/docker/services" path = f"/endpoints/{self.get_endpoint_id(endpoint)}/docker/services"
print(path) # print(path)
# path += f'?filters={{"label": ["com.docker.compose.project={stack}"]}}' # path += f'?filters={{"label": ["com.docker.compose.project={stack}"]}}'
services = self._api_get(path, timeout=timeout) services = self._api_get(path, timeout=timeout)
return services return services
@@ -399,7 +400,11 @@ class Portainer:
for s in self.all_data["webhooks"][endpoint]: for s in self.all_data["webhooks"][endpoint]:
stcs.append([s, self.all_data["webhooks"][endpoint][s]]) stcs.append([s, self.all_data["webhooks"][endpoint][s]])
else: else:
try:
stcs.append([stack, self.all_data["webhooks"][endpoint][stack]]) stcs.append([stack, self.all_data["webhooks"][endpoint][stack]])
except Exception as e:
print(f"Error: Stack {stack} not found on endpoint {endpoint}: {e}")
# input(stcs) # input(stcs)
def update(c): def update(c):
@@ -724,7 +729,6 @@ class Portainer:
} }
self._api_post_file(path, self.endpoint_id, stack, envs, file) self._api_post_file(path, self.endpoint_id, stack, envs, file)
def print_stacks(self, endpoint="all"): def print_stacks(self, endpoint="all"):
"""Print a table of stacks, optionally filtered by endpoint.""" """Print a table of stacks, optionally filtered by endpoint."""
stacks = self.get_stacks() stacks = self.get_stacks()
@@ -763,43 +767,40 @@ class Portainer:
all_services = self.get_services(self.get_endpoint_id(endpoint_id)) all_services = self.get_services(self.get_endpoint_id(endpoint_id))
service_tuples = [(s['ID'], s['Spec']['Name']) for s in all_services] service_tuples = [(s['ID'], s['Spec']['Name']) for s in all_services]
service_tuples = sorted(service_tuples, key=lambda x: x[1])
input(service_tuples) service_dict = dict(service_tuples)
if service_id == "all": # input(service_tuples)
if service_id is None:
service_id = self.all
#services = [(s["Id"], s["Name"]) for s in self.get_stacks(endpoint_id)] #services = [(s["Id"], s["Name"]) for s in self.get_stacks(endpoint_id)]
services.insert(0, ("__ALL__", "[Select ALL]")) service_tuples.insert(0, ("__ALL__", "[Select ALL]"))
service_tuples.insert(0, ("__ONLY_CHECK__", "[Check Only]"))
service_ids = checkboxlist_dialog( service_ids = checkboxlist_dialog(
title="Select one service", title="Select one service",
text="Choose a service:", text="Choose a service:",
values=services values=service_tuples
).run() ).run()
if service_ids == "__ALL__": elif service_id == "all":
pass service_ids = [s[0] for s in service_tuples if s[0] != "__ALL__" and s[0] != "__ONLY_CHECK__"]
print(service_ids) else:
service_dict = dict(service_ids) service_ids = [service_id]
services = self.get_services(self.endpoint_name, stack_id) if "__ONLY_CHECK__" in service_ids:
svc_name = service_dict.get(stack_id) pull = False
stack_svcs = [] else:
svc_menu = [] pull = False
for s in services: if "__ALL__" in service_ids:
try: service_ids = [s[0] for s in service_tuples if s[0] != "__ALL__" and s[0] != "__ONLY_CHECK__"]
if svc_name in s['Spec']['Name']:
stack_svcs.append([s['Version']['Index'], s['Spec']['Name']])
svc_menu.append([s['ID'], s['Spec']['Name']])
except KeyError as e:
print(e)
longest = 0
service_id = radiolist_dialog( for a in service_dict.items():
title="Select one service", # print(a[1])
text="Choose a service:", if len(a[1]) > longest:
values=svc_menu longest = len(a[1])
).run() #print(longest)
ok = "\033[92m✔\033[0m"
err = "\033[91m✖\033[0m"
"""Restart a service on an endpoint.""" for service_id in service_ids:
print("\033[?25l", end="")
print(f"{service_dict[service_id]:<{longest}} ", end="", flush=True)
path = f"/docker/{self.endpoint_id}/services/{service_id}/image_status?refresh=true" path = f"/docker/{self.endpoint_id}/services/{service_id}/image_status?refresh=true"
try: try:
@@ -807,9 +808,19 @@ class Portainer:
except ValueError as e: except ValueError as e:
print(f"Error restarting service: {e}") print(f"Error restarting service: {e}")
return [] return []
if resp['Status'] == "outdated": if resp['Status'] == "outdated":
self.restart_srv(service_id, True) if pull:
print(f"Service {service_id} : restarted") self.restart_srv(service_id, pull)
print(f"Service {service_dict[service_id]:<{longest}} : updated")
else:
print(f"\r\033[4m{service_dict[service_id]:<{longest}}\033[0m ", end="", flush=True)
#print(f"\033[4m{service_dict[service_id]:<{longest}} {err}\033[0m")
pass
print(err)
else:
print(ok)
print("\033[?25h", end="")
return True return True
def restart_srv(self,service_id, pool=False): def restart_srv(self,service_id, pool=False):

46
portainer.py
View File

@@ -71,7 +71,7 @@ def load_config(defaults=defaults):
cur_config = load_config(defaults) a = load_config(defaults)
# ENV_VARS = [ # ENV_VARS = [
# "PORTAINER_URL", # "PORTAINER_URL",
@@ -116,46 +116,13 @@ parser.add_argument(
default=None, default=None,
help="Service ID to limit service operations", help="Service ID to limit service operations",
) )
parser.add_argument( parser.add_argument("--stack", "-s", type=str, nargs="+", help="Stack ID for operations")
"--refresh-environment", "-R", action="store_true", help="List endpoints"
)
parser.add_argument(
"--list-endpoints", "-E", action="store_true", help="List endpoints"
)
parser.add_argument("--list-stacks", "-l", action="store_true", help="List stacks")
parser.add_argument("--print-all-data", "-A", action="store_true", help="List stacks")
parser.add_argument(
"--list-containers", "-c", action="store_true", help="List containers"
)
parser.add_argument("--update-stack", "-U", action="store_true", help="Update stacks")
parser.add_argument(
"--stop-containers", "-O", action="store_true", help="Stop containers"
)
parser.add_argument(
"--start-containers", "-X", action="store_true", help="Start containers"
)
parser.add_argument("--update-status", "-S", action="store_true", help="Update status")
parser.add_argument(
"--get-stack", metavar="NAME_OR_ID", help="Get stack by name or numeric id"
)
parser.add_argument("--action", "-a", type=str, default=None, help="Action to perform") parser.add_argument("--action", "-a", type=str, default=None, help="Action to perform")
parser.add_argument( parser.add_argument(
"--autostart", "-Z", action="store_true", help="Auto-start created stacks" "--autostart", "-Z", action="store_true", help="Auto-start created stacks"
) )
parser.add_argument("--start-stack", "-x", action="store_true")
parser.add_argument("--stop-stack", "-o", action="store_true")
parser.add_argument("--secrets", "-q", action="store_true")
parser.add_argument("--debug", "-D", action="store_true") parser.add_argument("--debug", "-D", action="store_true")
parser.add_argument("--create-stack", "-n", action="store_true")
parser.add_argument("--create-stack_new2", "-N", action="store_true")
parser.add_argument("--gpu", "-g", action="store_true") parser.add_argument("--gpu", "-g", action="store_true")
parser.add_argument("--create-stacks", "-C", action="store_true")
parser.add_argument("--refresh-status", "-r", action="store_true")
parser.add_argument("--stack", "-s", type=str, nargs="+", help="Stack ID for operations")
parser.add_argument(
"--token-only", action="store_true", help="Print auth token and exit"
)
parser.add_argument("--timeout", type=int, default=10, help="Request timeout seconds") parser.add_argument("--timeout", type=int, default=10, help="Request timeout seconds")
parser.add_argument("--deploy-mode", "-m", type=str, default="git", help="Deploy mode") parser.add_argument("--deploy-mode", "-m", type=str, default="git", help="Deploy mode")
parser.add_argument("--stack-mode", "-w", default=None, help="Stack mode") parser.add_argument("--stack-mode", "-w", default=None, help="Stack mode")
@@ -235,10 +202,13 @@ def prompt_missing_args(args_in, defaults_in, fields, action=None,stacks=None):
longest = len(a) longest = len(a)
for field, text in fields: for field, text in fields:
# print(field)
value_in = getattr(args_in, field) value_in = getattr(args_in, field)
default = defaults_in.get(f"PORTAINER_{field}".upper()) default = defaults_in.get(f"PORTAINER_{field}".upper())
cur_site = defaults_in.get("PORTAINER_SITE".upper()) cur_site = defaults_in.get("PORTAINER_SITE".upper())
cur_env = defaults_in.get("PORTAINER_ENVIRONMENT_ID".upper()) cur_env = defaults_in.get("PORTAINER_ENVIRONMENT_ID".upper())
# print(value_in)
if value_in is None: if value_in is None:
if default is not None: if default is not None:
prompt_text = f"{text} (default={default}) : " prompt_text = f"{text} (default={default}) : "
@@ -390,6 +360,7 @@ if __name__ == "__main__":
def signal_handler(sig, frame): def signal_handler(sig, frame):
logger.warning("Killed manually %s, %s", sig, frame) logger.warning("Killed manually %s, %s", sig, frame)
print("\nTerminated by user") print("\nTerminated by user")
print("\033[?25h", end="")
sys.exit(0) sys.exit(0)
signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGINT, signal_handler)
os.system("cls" if os.name == "nt" else "clear") os.system("cls" if os.name == "nt" else "clear")
@@ -438,7 +409,7 @@ if __name__ == "__main__":
os.system("cls" if os.name == "nt" else "clear") os.system("cls" if os.name == "nt" else "clear")
# Example: list endpoints # Example: list endpoints
por = Portainer(cur_config["PORTAINER_SITE"], timeout=args.timeout) por = Portainer(cur_config["PORTAINER_SITE"], args)
por.set_defaults(cur_config) por.set_defaults(cur_config)
if args.debug: if args.debug:
por._debug = True por._debug = True
@@ -548,6 +519,9 @@ if __name__ == "__main__":
sys.exit() sys.exit()
if args.action == "update_service": if args.action == "update_service":
args = prompt_missing_args( args = prompt_missing_args(
args, args,
cur_config, cur_config,