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-11-30 18:01:10 +01:00
parent b65680be59
commit 3595f832f2
2 changed files with 49 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
port.py
View File

@@ -51,16 +51,14 @@ class Portainer:
return True
except ValueError:
return False
def api_get(self, path, timeout=120):
url = f"{self.base_url.rstrip('/')}{path}"
headers = {"X-API-Key": f"{self.token}"}
resp = requests.get(url, headers=headers, timeout=timeout)
resp.raise_for_status()
return resp.json()
def api_post(self, path, json="", timeout=120):
url = f"{self.base_url.rstrip('/')}{path}"
headers = {"X-API-Key": f"{self.token}"}
@@ -92,7 +90,6 @@ class Portainer:
resp = requests.post(url, headers=headers, timeout=timeout)
return resp.text
def api_delete(self, path, timeout=120):
"""Example authenticated DELETE request to Portainer API."""
url = f"{self.base_url.rstrip('/')}{path}"
@@ -102,16 +99,12 @@ class Portainer:
resp.raise_for_status()
#print(resp.status_code)
return resp.status_code
# Higher-level operations
def refresh(self):
self.get_endpoints()
self.get_stacks(self)
self.get_containers(self)
def get_stacks(self, endpoint_id="all", timeout=10):
if endpoint_id != "all":
endpoint_id = self.get_endpoint_id(endpoint_id)
@@ -143,14 +136,15 @@ class Portainer:
self.stacks_all[self.endpoints["by_id"][s['EndpointId']]]["by_name"][s['Name']] = s['Id']
#print(s)
if type(s["AutoUpdate"]) == dict and "Webhook" in s["AutoUpdate"]:
#print(self.endpoints["by_id"][s['EndpointId']], s['Name'], s["AutoUpdate"]['Webhook'])
#print("WWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWW")
webhooks[s['EndpointId']][s['Name']] = s['AutoUpdate']['Webhook']
webhooks[self.endpoints["by_id"][s['EndpointId']]][s['Name']] = s['AutoUpdate']['Webhook']
elif s["Webhook"] != "":
webhooks[s['EndpointId']][s['Name']] = s['Webhook']
webhooks[self.endpoints["by_id"][s['EndpointId']]][s['Name']] = s['Webhook']
if "AutoUpdate" in s and s["AutoUpdate"] is not None:
if type(s["AutoUpdate"]) == dict and "Webhook" in s["AutoUpdate"]:
#print(self.endpoints["by_id"][s['EndpointId']], s['Name'], s["AutoUpdate"]['Webhook'])
#print("WWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWW")
webhooks[s['EndpointId']][s['Name']] = s['AutoUpdate']['Webhook']
webhooks[self.endpoints["by_id"][s['EndpointId']]][s['Name']] = s['AutoUpdate']['Webhook']
elif s["AutoUpdate"]["Webhook"] != "":
webhooks[s['EndpointId']][s['Name']] = s['Webhook']
webhooks[self.endpoints["by_id"][s['EndpointId']]][s['Name']] = s['Webhook']
#print(self.stacks_all)
if s['EndpointId'] == endpoint_id or endpoint_id == "all":
@@ -163,13 +157,16 @@ class Portainer:
self.all_data["webhooks"] = webhooks
#input(json.dumps(self.stacks_all,indent=2))
return stcks
def get_stack_id(self,endpoint,stack):
pass
def update_status(self,endpoint,stack):
path = f"/stacks/{self.all_data['stacks'][endpoint]['by_name'][stack]}/images_status?refresh=true"
#input(path)
stats = self.api_get(path)
print(stats)
def get_endpoint_id(self,endpoint):
if self.is_number(endpoint):
self.endpoint_id = endpoint
@@ -236,7 +233,6 @@ class Portainer:
return cont
def stop_containers(self, endpoint, containers, timeout=130):
if self.all_data["endpoints_status"][endpoint] != 1:
print(f"Endpoint {self.get_endpoint_name(endpoint)} is offline")
@@ -265,6 +261,7 @@ class Portainer:
)
with ThreadPoolExecutor(max_workers=10) as exe:
exe.map(stop, containers)
def update_stack(self, endpoint, stack, autostart, timeout=130):
stcs = []
if stack == "all":
@@ -296,8 +293,7 @@ class Portainer:
if stack == c or stack== "all":
cont+=self.all_data["containers"][endpoint][c]
self.stop_containers(endpoint,cont)
def get_endpoints(self, timeout=10):
endpoints = self.api_get("/endpoints")
eps = {"by_id":{}, "by_name":{}}
@@ -325,10 +321,12 @@ class Portainer:
self.endpoint_id = self.endpoints["by_name"][endpoint_id]
return self.endpoint_id
def get_swarm_id(self, endpoint):
ep_id = self.endpoints["by_name"][endpoint]
path = f"/endpoints/{ep_id}/docker/info"
stats = self.api_get(path)
return stats['Swarm']['Cluster']['ID']
def get_stack(self, stack=None, endpoint_id=None, timeout=None):
self.get_stacks(endpoint_id)
@@ -351,8 +349,9 @@ class Portainer:
return s
raise ValueError(f"Stack not found: {stack}")
def create_stack(self, endpoint, stack=None, mode="git", autostart=False, timeout=None):
def create_stack(self, endpoint, stack=None, mode="git", autostart=False, ep_mode='swarm', timeout=None):
swarm_id = self.get_swarm_id(endpoint)
#input(swarm_id)
self.endpoint_id = self.get_endpoint_id(endpoint)
if os.path.exists(self.repo_dir):
shutil.rmtree(self.repo_dir)
@@ -364,9 +363,12 @@ class Portainer:
print("Creating new stack from git repo...")
enviro="swarm"
path = f"/stacks/create/{enviro}/repository"
if self.endpoint_id is not None:
path += f"?endpointId={self.endpoint_id}"
print(path)
if stack == "all":
if self.endpoint_name == "rack":
stacks = self.rack_stacks
@@ -386,8 +388,8 @@ class Portainer:
print(f"Stack {stack} already exist")
continue
print(f"Working on {stack}")
if os.path.exists(f"{self.repo_dir}/{stack}/.env"):
f = open(f"{self.repo_dir}/{stack}/.env","r")
if os.path.exists(f"{self.repo_dir}/__swarm/{stack}/.env"):
f = open(f"{self.repo_dir}/__swarm/{stack}/.env","r")
env_vars = f.read().splitlines()
envs = []
for ev in env_vars:
@@ -447,10 +449,15 @@ class Portainer:
"filesystemPath": "/share/docker_data/portainer/portainer-data/",
"RegistryID": 4,
"isDetachedFromGit": True,
"swarmID": "9mqis14p7bogfhf7swndwjqja",
"method":"repository",
"type":"swarm"
"method":"repository"
}
if ep_mode == "swarm":
req["type"] = "swarm"
req["swarmID"] = swarm_id
req["composeFile"] = f"__swarm/{stack}/{stack}-swarm.yml"
req["ConfigFilePath"] = f"__swarm/{stack}/{stack}-swarm.yml"
print(json.dumps(req))
res = self.api_post(path,req)
if "Id" in res:
#print("Deploy request OK")
@@ -580,6 +587,7 @@ class Portainer:
else:
print(f"Stack {self.stacks_all[self.endpoint_id]['by_id'][stack]} : {json.loads(resp)['message']}")
return True
def stop_stack(self,stack,endpoint_id):
print(f"Stopping stack {stack}")
if endpoint_id != None:

10
portainer.py
View File

@@ -13,6 +13,7 @@ import logging
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"),
help="Base URL for Portainer (ENV: PORTAINER_URL)")
parser.add_argument("--site", "-t", type=str, default="portainer", help="Site")
parser.add_argument("--endpoint-id", "-e", type=str, default="all", help="Endpoint ID to limit stack operations")
parser.add_argument("--refresh-environment", "-R", action="store_true", help="List endpoints")
parser.add_argument("--list-endpoints","-E", action="store_true", help="List endpoints")
@@ -56,7 +57,13 @@ logging.info("script started")
logger = logging.getLogger(__name__)
portainer_api_key = "ptr_GCNUoFcTOaXm7k8ZxPdQGmrFIamxZPTydbserYofMHc="
if args.site == "portainer":
base = os.getenv("PORTAINER_URL", "https://portainer.sectorq.eu/api")
portainer_api_key = "ptr_GCNUoFcTOaXm7k8ZxPdQGmrFIamxZPTydbserYofMHc="
else:
base = os.getenv("PORTAINER_URL", "https://port.sectorq.eu/api")
portainer_api_key = "ptr_/5RkMCT/j3BTaL32vMSDtXFi76yOXRKVFOrUtzMsl5Y="
def wl(msg):
if args.debug:
print(msg)
@@ -91,7 +98,6 @@ def get_portainer_token(base_url, username=None, password=None, timeout=10):
if __name__ == "__main__":
# Example usage: set PORTAINER_USER and PORTAINER_PASS in env, or pass literals below.
base = os.getenv("PORTAINER_URL", "https://portainer.sectorq.eu/api")
#token = get_portainer_token(base,"admin","l4c1j4yd33Du5lo") # or get_portainer_token(base, "admin", "secret")
token = portainer_api_key
# Example: list endpoints