build

portainer.py

@@ -0,0 +1,173 @@
+import os
+import requests
+import json
+import uuid
+# portainer.py
+
+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 api_get(base_url, path, token, timeout=10):
+    """Example authenticated GET request to Portainer API."""
+    url = f"{base_url.rstrip('/')}{path}"
+    headers = {"Authorization": f"Bearer {token}"}
+    resp = requests.get(url, headers=headers, timeout=timeout)
+    resp.raise_for_status()
+    return resp.json()
+
+def api_post(base_url, path, token, data, timeout=20):
+    """Example authenticated GET request to Portainer API."""
+    url = f"{base_url.rstrip('/')}{path}"
+    headers = {"Authorization": f"Bearer {token}"}
+    resp = requests.post(url, headers=headers, json=data, timeout=timeout)
+    resp.raise_for_status()
+    return resp.json()
+
+
+def get_stacks(base_url, token, endpoint_id=None, timeout=10):
+    """
+    Return a list of stacks. If endpoint_id is provided, it will be added as a query param.
+    """
+    path = "/api/stacks"
+    if endpoint_id is not None:
+        path += f"?endpointId={endpoint_id}"
+    stacks = api_get(base_url, path, token, timeout=timeout)
+    if stacks is None:
+        return []
+    return stacks
+
+def get_stack(base_url, identifier, token, endpoint_id=None, timeout=10):
+    """
+    Retrieve a single stack by numeric Id or by Name.
+    Identifier may be an int (Id) or a string (Name). Raises ValueError if not found.
+    """
+    stacks = get_stacks(base_url, token, endpoint_id=endpoint_id, timeout=timeout)
+    # Normalize identifier
+    ident_id = None
+    try:
+        ident_id = int(identifier)
+    except (TypeError, ValueError):
+        pass
+
+    for s in stacks:
+        # Many Portainer responses use 'Id' and 'Name' keys
+        if ident_id is not None and s.get("Id") == ident_id:
+            return s
+        if str(s.get("Name")) == str(identifier):
+            return s
+
+    raise ValueError(f"Stack not found: {identifier}")
+
+def create_stack(base_url, token, endpoint_id=None, data={}, timeout=10):
+    """
+    Return a list of stacks. If endpoint_id is provided, it will be added as a query param.
+    """
+    path = "/api/stacks/create/standalone/repository"
+    if endpoint_id is not None:
+        path += f"?endpointId={endpoint_id}"
+    stacks = api_post(base_url, path, token, data, timeout=timeout)
+    if stacks is None:
+        return []
+    return stacks
+
+
+
+
+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")
+    token = get_portainer_token(base,"admin","l4c1j4yd33Du5lo")  # or get_portainer_token(base, "admin", "secret")
+    # Example: list endpoints
+    endpoints = api_get(base, "/api/endpoints", token)
+    #print(endpoints)
+    for ep in endpoints:
+        print(f"Endpoint ID: {ep['Id']}, Name: {ep['Name']}")
+
+
+    stacks = get_stacks(base, token)
+    for stack in stacks:
+        print(f"Stack ID: {stack['Id']}, Name: {stack['Name']}")
+
+    stck = get_stack(base, "pihole", token)
+
+
+
+    print(f"Found stack: ID {stck['Id']}, Name: {stck['Name']}")
+    print(json.dumps(stck, indent=2))
+
+    uid = str(uuid.uuid4())
+    try:
+        stck["AutoUpdate"]["Webhook"] = uid
+    except:
+        stck["AutoUpdate"] = {}
+    try:
+        req = {
+            "Name": stck["Name"],
+            "Env": stck["Env"],
+            "AdditionalFiles": stck["AdditionalFiles"],
+            "AutoUpdate": stck["AutoUpdate"],
+            "repositoryURL": stck["GitConfig"]["URL"],
+            "ReferenceName": "refs/heads/main",
+            "composeFile": f"{stck['Name']}/docker-compose.yml",
+            "ConfigFilePath": f"{stck['Name']}/docker-compose.yml",
+            "repositoryAuthentication": True,
+            "repositoryUsername": "jaydee",
+            "repositoryPassword": "glpat-uj-n-eEfTY398PE4vKSS",
+            "AuthorizationType": 0,
+            "TLSSkipVerify": False,
+            "supportRelativePath": True,
+            "repositoryAuthentication": True,
+            "fromAppTemplate": False,
+            "registries": [],
+            "FromAppTemplate": False,
+            "Namespace": "",
+            "CreatedByUserId": "",
+            "Webhook": "",
+            "filesystemPath": "/tmp",
+            "RegistryID": 6
+        }
+    except:
+        req = {
+        "Name": stck["Name"],
+        "Env": stck["Env"],
+        "AdditionalFiles": stck["AdditionalFiles"],
+        "AutoUpdate": None,
+        "repositoryURL": "https://gitlab.sectorq.eu/home/docker-compose.git",
+        "ReferenceName": "refs/heads/main",
+        "composeFile": f"{stck['Name']}/docker-compose.yml",
+        "ConfigFilePath": f"{stck['Name']}/docker-compose.yml",
+        "repositoryAuthentication": True,
+        "repositoryUsername": "jaydee",
+        "repositoryPassword": "glpat-uj-n-eEfTY398PE4vKSS",
+        "AuthorizationType": 0,
+        "TLSSkipVerify": False,
+        "supportRelativePath": True,
+        "repositoryAuthentication": True,
+        "fromAppTemplate": False,
+        "registries": [],
+        "FromAppTemplate": False,
+        "Namespace": "",
+        "CreatedByUserId": "",
+        "Webhook": "",
+        "filesystemPath": "/tmp",
+        "RegistryID": 6
+    }
+    print(json.dumps(req, indent=2))
+
+    create_stack(base, token, 41, data=req)