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

build

Browse Source
This commit is contained in:
Ladislav Dusa
2025-12-12 15:19:53 +01:00
parent 5864085ec3
commit d6842eab62
2 changed files with 29 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
port.py
View File

@@ -146,6 +146,22 @@ class Portainer:
except ValueError:
return False
def gotify_message(self, message):
payload = {
"title": "Updates in Portainer",
"message": message,
"priority": 5
}
'''Send a notification message via Gotify.'''
response = requests.post(
"https://gotify.sectorq.eu/message",
data=payload,
headers={"X-Gotify-Key": "ASn_fIAd5OVjm8c"}
)
# print("Status:", response.status_code)
# print("Response:", response.text)
pass
def _api_get(self, path, timeout=120):
url = f"{self.base_url.rstrip('/')}{path}"
headers = {"X-API-Key": f"{self.token}"}
@@ -763,14 +779,14 @@ class Portainer:
print(f"Total stacks: {count}")
# print(sorted(stack_names))
def update_service(self, endpoint_id=None, service_id=None):
all_services = self.get_services(self.get_endpoint_id(endpoint_id))
def update_service(self):
all_services = self.get_services(self.get_endpoint_id(self.args.endpoint_id))
service_tuples = [(s['ID'], s['Spec']['Name']) for s in all_services]
service_tuples = sorted(service_tuples, key=lambda x: x[1])
service_dict = dict(service_tuples)
# input(service_tuples)
if service_id is None:
if self.args.service_id is None:
#services = [(s["Id"], s["Name"]) for s in self.get_stacks(endpoint_id)]
service_tuples.insert(0, ("__ALL__", "[Select ALL]"))
service_tuples.insert(0, ("__ONLY_CHECK__", "[Check Only]"))
@@ -779,11 +795,11 @@ class Portainer:
text="Choose a service:",
values=service_tuples
).run()
elif service_id == "all":
elif self.args.service_id == "all":
service_ids = [s[0] for s in service_tuples if s[0] != "__ALL__" and s[0] != "__ONLY_CHECK__"]
else:
service_ids = [service_id]
if "__ONLY_CHECK__" in service_ids:
service_ids = [self.args.service_id]
if "__ONLY_CHECK__" in service_ids and self.args.update is False:
pull = False
else:
pull = True
@@ -812,11 +828,13 @@ class Portainer:
if resp['Status'] == "outdated":
if pull:
self.restart_srv(service_id, pull)
print(f"Service {service_dict[service_id]:<{longest}} : updated")
#print(f"Service {service_dict[service_id]:<{longest}} : updated")
self.gotify_message(f"Service {service_dict[service_id]} updated")
print(ok)
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
self.gotify_message(f"Service update available for {service_dict[service_id]}")
print(err)
else:
print(ok)

4
portainer.py
View File

@@ -121,6 +121,7 @@ parser.add_argument("--action", "-a", type=str, default=None, help="Action to pe
parser.add_argument(
"--autostart", "-Z", action="store_true", help="Auto-start created stacks"
)
parser.add_argument("--update", "-u", action="store_true", help="Update service if it exists")
parser.add_argument("--debug", "-D", action="store_true")
parser.add_argument("--gpu", "-g", action="store_true")
parser.add_argument("--timeout", type=int, default=10, help="Request timeout seconds")
@@ -353,7 +354,6 @@ def prompt_missing_args(args_in, defaults_in, fields, action=None,stacks=None):
return args
if __name__ == "__main__":
# 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")
@@ -530,7 +530,7 @@ if __name__ == "__main__":
("endpoint_id", "Endpoint ID")
],
)
por.update_service(args.endpoint_id, args.service_id)
por.update_service()
sys.exit()
if args.action == "list_stacks":
args = prompt_missing_args(