build

port.py

@@ -163,6 +163,7 @@ class Portainer:
         self.get_endpoints()
         self.get_stacks(self)
         self.get_containers(self)
+        return True
 
     def get_stacks(self, endpoint_id="all", timeout=10):
         if endpoint_id != "all":
@@ -819,4 +820,4 @@ class Portainer:
         path = f"/endpoints/{endpoint_id}/docker/secrets/create"
         encoded = base64.b64encode(value.encode()).decode()
         data = {"Name": name, "Data": encoded}
-        self.api_post(path, data, timeout=timeout)
+        return self.api_post(path, data, timeout=timeout)

portainer.py

@@ -1,12 +1,19 @@
+"""
+Portainer API Client module.
+
+This module provides a wrapper for interacting with the Portainer API
+to manage endpoints, stacks, and containers.
+"""
+
 #!/myapps/venvs/portainer/bin/python3
 import os
+import logging
 import sys
-import requests
 import json
 import argparse
 from tabulate import tabulate
 from port import Portainer
-import logging
+
 
 VERSION = "0.0.1"
 
@@ -123,13 +130,14 @@ logger = logging.getLogger(__name__)
 
 if args.site == "portainer":
     base = os.getenv("PORTAINER_URL", "https://portainer.sectorq.eu/api")
-    portainer_api_key = "ptr_GCNUoFcTOaXm7k8ZxPdQGmrFIamxZPTydbserYofMHc="
+    PORTAINER_API_KEY = "ptr_GCNUoFcTOaXm7k8ZxPdQGmrFIamxZPTydbserYofMHc="
 else:
     base = os.getenv("PORTAINER_URL", "https://port.sectorq.eu/api")
-    portainer_api_key = "ptr_/5RkMCT/j3BTaL32vMSDtXFi76yOXRKVFOrUtzMsl5Y="
+    PORTAINER_API_KEY = "ptr_/5RkMCT/j3BTaL32vMSDtXFi76yOXRKVFOrUtzMsl5Y="
 
 
 def wl(msg):
+    """Write log message if debug is enabled."""
     if args.debug:
         print(msg)
 
@@ -143,46 +151,22 @@ def is_number(s):
         return False
 
 
-def get_portainer_token(base_url, username=None, password=None, timeout=10):
-    """
-    Authenticate to Portainer and return a JWT token.
-    Reads PORTAINER_USER / PORTAINER_PASS from environment if username/password are not provided.
-    """
-    username = username or os.getenv("PORTAINER_USER")
-    password = password or os.getenv("PORTAINER_PASS")
-    if not username or not password:
-        raise ValueError(
-            "Username and password must be provided (or set PORTAINER_USER / PORTAINER_PASS)."
-        )
-
-    url = f"{base_url.rstrip('/')}/api/auth"
-    resp = requests.post(
-        url, json={"Username": username, "Password": password}, timeout=timeout
-    )
-    resp.raise_for_status()
-    data = resp.json()
-    token = data.get("jwt") or data.get("JWT") or data.get("token")
-    if not token:
-        raise ValueError(f"No token found in response: {data}")
-    return token
-
-
-def prompt_missing_args(args, defaults, fields):
+def prompt_missing_args(args_in, defaults_in, fields):
     """
     fields = [("arg_name", "Prompt text")]
     """
     for field, text in fields:
-        value = getattr(args, field)
-        default = defaults.get(field)
+        value_in = getattr(args_in, field)
+        default = defaults_in.get(field)
 
-        if value is None:
+        if value_in is None:
             if default is not None:
                 prompt = f"{text} (default={default}) : "
-                value = input(prompt) or default
+                value_in = input(prompt) or default
             else:
-                value = input(f"{text}: ")
+                value_in = input(f"{text}: ")
 
-            setattr(args, field, value)
+            setattr(args, field, value_in)
 
     return args
 
@@ -216,9 +200,8 @@ if __name__ == "__main__":
         ans = input("\nSelect action to perform: ")
         args.action = actions[int(ans) - 1]
 
-    token = portainer_api_key
     # Example: list endpoints
-    por = Portainer(base, token)
+    por = Portainer(base, PORTAINER_API_KEY, timeout=args.timeout)
 
     if args.action == "secrets":
         if args.endpoint_id is None:
@@ -301,8 +284,7 @@ if __name__ == "__main__":
 
     if args.action == "update_stack":
         print("Updating stacks")
-        autostart = True if args.autostart else False
-        por.update_stack(args.endpoint_id, args.stack, autostart)
+        por.update_stack(args.endpoint_id, args.stack, args.autostart)
         sys.exit()
     if args.action == "print_all_data":
         print(json.dumps(por.all_data, indent=2))
@@ -313,11 +295,11 @@ if __name__ == "__main__":
 
     if args.action == "list_endpoints":
         eps = por.get_endpoints()
-        data = []
+        export_data = []
         for i in eps["by_id"]:
-            data.append([i, eps["by_id"][i]])
+            export_data.append([i, eps["by_id"][i]])
         headers = ["EndpointId", "Name"]
-        print(tabulate(data, headers=headers, tablefmt="github"))
+        print(tabulate(export_data, headers=headers, tablefmt="github"))
 
         sys.exit()
 
@@ -328,7 +310,7 @@ if __name__ == "__main__":
         print(f"Stopping containers on {por.get_endpoint_name(args.endpoint_id)}")
         cont = []
         for c in por.all_data["containers"][args.endpoint_id]:
-            if args.stack == c or args.stack == "all":
+            if args.stack in (c, "all"):
                 cont += por.all_data["containers"][args.endpoint_id][c]
         por.stop_containers(args.endpoint_id, cont)
         sys.exit()
@@ -338,7 +320,7 @@ if __name__ == "__main__":
         cont = []
         # input(json.dumps(por.all_data, indent=2))
         for c in por.all_data["containers"][args.endpoint_id]:
-            if args.stack == c or args.stack == "all":
+            if args.stack in (c, "all"):
                 cont += por.all_data["containers"][args.endpoint_id][c]
         por.start_containers(args.endpoint_id, cont)
         sys.exit()
@@ -347,7 +329,7 @@ if __name__ == "__main__":
         cont = []
         # input(json.dumps(por.all_data,indent=2))
         for c in por.all_data["containers"][args.endpoint_id]:
-            if args.stack == c or args.stack == "all":
+            if args.stack in (c, "all"):
                 cont += por.all_data["containers"][args.endpoint_id][c]
         por.start_containers(args.endpoint_id, cont)
         sys.exit()
@@ -358,6 +340,6 @@ if __name__ == "__main__":
     if args.action == "refresh_status":
         if args.stack == "all":
             print("Stopping all stacks...")
-            stcks = por.get_stacks(base, token, endpoint_id=args.endpoint_id)
+            stcks = por.get_stacks(endpoint_id=args.endpoint_id)
         else:
-            por.refresh_status(base, args.stack_id, token)
+            por.refresh_status(args.stack_id)