portainer_host: implement list()

services/tools/cli/api.py

@@ -19,7 +19,6 @@ class ApiClient():
             token = self.default_token
         if token:
             headers['Authorization'] = f'Bearer {token}'
-
         try:
             if method.upper() == 'POST':
                 response = requests.post(request_url, json=payload, headers=headers)
@@ -33,7 +32,7 @@ class ApiClient():
                 raise ValueError("Unsupported HTTP method: {}".format(method))
 
             if response.status_code != 200:
-                self.warning(f'Return code {response.status_code} from {endpoint} with message {response.text}')
+                self.warning(f'Return code {response.status_code} from {request_url} with request {method} with message {response.text}')
                 
         except requests.exceptions.RequestException as e:
             self.error("An error occurred:", e)

services/tools/cli/dataclass/portainer/stack.py

@@ -0,0 +1,87 @@
+from dataclasses import dataclass, field, fields
+from typing import List, Optional
+
+@dataclass
+class Env:
+    name: Optional[str] = None
+    value: Optional[str] = None
+
+@dataclass
+class TeamAccess:
+    AccessLevel: Optional[int] = None
+    TeamId: Optional[int] = None
+
+@dataclass
+class UserAccess:
+    AccessLevel: Optional[int] = None
+    UserId: Optional[int] = None
+
+@dataclass
+class ResourceControl:
+    AccessLevel: Optional[int] = None
+    AdministratorsOnly: Optional[bool] = None
+    Id: Optional[int] = None
+    OwnerId: Optional[int] = None
+    Public: Optional[bool] = None
+    ResourceId: Optional[str] = None
+    SubResourceIds: List[str] = field(default_factory=list)
+    System: Optional[bool] = None
+    TeamAccesses: List[TeamAccess] = field(default_factory=list)
+    Type: Optional[int] = None
+    UserAccesses: List[UserAccess] = field(default_factory=list)
+
+@dataclass
+class AutoUpdate:
+    forcePullImage: Optional[bool] = None
+    forceUpdate: Optional[bool] = None
+    interval: Optional[str] = None
+    jobID: Optional[str] = None
+    webhook: Optional[str] = None
+
+@dataclass
+class GitAuthentication:
+    gitCredentialID: Optional[int] = None
+    password: Optional[str] = None
+    username: Optional[str] = None
+
+@dataclass
+class GitConfig:
+    authentication: GitAuthentication = GitAuthentication()
+    configFilePath: Optional[str] = None
+    configHash: Optional[str] = None
+    referenceName: Optional[str] = None
+    tlsskipVerify: Optional[bool] = None
+    url: Optional[str] = None
+
+@dataclass
+class Option:
+    prune: Optional[bool] = None
+
+@dataclass
+class PortainerStack:
+    AdditionalFiles: List[str] = field(default_factory=list)
+    AutoUpdate: AutoUpdate = AutoUpdate()
+    EndpointId: Optional[int] = None
+    EntryPoint: Optional[str] = None
+    Env: List['Env'] = field(default_factory=list)
+    Id: Optional[int] = None
+    Name: Optional[str] = None
+    Option: Option = Option()
+    ResourceControl: ResourceControl = ResourceControl()
+    Status: Optional[int] = None
+    SwarmId: Optional[str] = None
+    Type: Optional[int] = None
+    createdBy: Optional[str] = None
+    creationDate: Optional[int] = None
+    fromAppTemplate: Optional[bool] = None
+    gitConfig: GitConfig = GitConfig()
+    namespace: Optional[str] = None
+    projectPath: Optional[str] = None
+    updateDate: Optional[int] = None
+    updatedBy: Optional[str] = None
+    
+def classFromArgs(className, argDict):
+    fieldSet = {f.name for f in fields(className) if f.init}
+    filteredArgDict = {k : v for k, v in argDict.items() if k in fieldSet}
+    return className(**filteredArgDict)
+

services/tools/cli/portainer_host.py

@@ -6,7 +6,9 @@ from constants import (
     FUNCTIONAL_USER_USERNAME_ENV,
     FUNCTIONAL_USER_PASSWORD_ENV,
 )
+from dataclass.portainer.stack import PortainerStack, classFromArgs
 import os
+from typing import List
 import requests
 
 class MissingEnvironmentVariable(Exception):
@@ -40,7 +42,8 @@ class PortainerHost(BaseCLi):
             print(f"Configuration does not exist. Creating...")
             raise EmptyConfigException()
         try:
-            self.jwt_token = self.jwt_config[self.host_url]
+            jwt_token = self.jwt_config[self.host_url]['jwt']
+            self.api.reassign_token(jwt_token)
         except KeyError:
             print(f"No token present in configuration file for host {self.host_url}. Creating...")
             raise EmptyConfigException()
@@ -72,6 +75,12 @@ class PortainerHost(BaseCLi):
         except KeyError:
             self.error('No "jwt" key in /auth response found.')
         return token
+
+    def list(self) -> List[PortainerStack]:
+        response = self.api.get('/stacks')
+        raw_stacks = response.json()
+        return [classFromArgs(PortainerStack, stack) for stack in raw_stacks]
+
     
         
 

services/tools/cli/portainer_stack.py

@@ -69,6 +69,7 @@ class PortainerStackCLi(BaseCLi):
 
     def start(self):
         host = PortainerHost(PORTAINER_HOST_URLS[self.args.hostname])
+        print(host.list()[0])
         
 
 def main():