initial implementation of sdk methods, req path changes

swan/api/swan_api.py

@@ -1,50 +1,107 @@
 import web3
+import logging
 
 from swan.api_client import APIClient
+from swan.common.constant import GET, POST
+
 
 class SwanAPI(APIClient):
 
-    def __init__(self, orchestrator_url, api_key, payment_key): 
-        self.orchestrator_url = orchestrator_url 
-        self.api_key = api_key 
+    def __init__(self, orchestrator_url, api_key, payment_key):
+        super().__init__(api_key)
+        self.orchestrator_url = orchestrator_url
+        self.api_key = api_key
         self.payment_key = payment_key
-
-    def query_price_list(self): 
-        """Query the orchestrator for the current instance price list. 
-        """ 
-        pass 
-
-    def build_task(self, source_code_url, instance_type, task_name, public_key=None): 
-        """Prepare a task for deployment with the required details. 
-        - source_code_url: URL to the code repository containing Docker/K8s file and env file 
-        - instance_type: Type of instance needed for the task 
-        - task_name: A name for the task 
-        - public_key: Optional public key for accessing confidential data 
-        """ 
-        pass 
-
-    def propose_task(self): 
-        """Propose the prepared task to the orchestrator. 
-        """ 
-        pass 
-
-    def make_payment(self): 
-        """Make payment for the task build after acceptance by the orchestrator. 
-        """ 
-        pass 
-
-    def get_payment_info(self): 
-        """Retrieve payment information from the orchestrator after making the payment. 
-        """ 
-        pass 
-
-    def get_task_status(self): 
-        """Fetch the current status of the task from the orchestrator. 
-        """ 
-        pass 
-
-    def fetch_task_details(self): 
-        """Retrieve the deployed URL and credentials/token for access after task success. 
-        Decrypt the credentials/token with the private key if necessary. 
-        """ 
-        pass 
+
+    def query_price_list(self):
+        """Query the orchestrator for the current instance price list."""
+        try:
+            price_list = self._request_without_params(
+                GET, PRICE_LIST, self.orchestrator_url, self.token
+            )
+            return price_list
+        except:
+            logging.error("An error occurred while executing query_price_list()")
+            return None
+
+    def build_task(self, source_code_url, instance_type, task_name, public_key=None):
+        """Prepare a task for deployment with the required details.
+        - source_code_url: URL to the code repository containing Docker/K8s file and env file
+        - instance_type: Type of instance needed for the task
+        - task_name: A name for the task
+        - public_key: Optional public key for accessing confidential data
+        """
+        try:
+            params = {
+                "source_code_url": source_code_url,
+                "instance_type": instance_type,
+                "task_name": task_name,
+                "public_key": public_key,
+            }
+            result = self._request_with_params(
+                POST, BUILD_TASK, self.orchestrator_url, params, self.token
+            )
+            return result
+        except:
+            logging.error("An error occurred while executing build_task()")
+            return None
+
+    def propose_task(self):
+        """Propose the prepared task to the orchestrator."""
+        try:
+            result = self._request_with_params(
+                POST, PROPOSE_TASK, self.orchestrator_url, {}, self.token
+            )
+            return result
+        except Exception as e:
+            return None
+
+    def make_payment(self):
+        """Make payment for the task build after acceptance by the orchestrator."""
+        try:
+            payment_info = self._request_without_params(
+                GET, PAYMENT_INFO, self.orchestrator_url, self.token
+            )
+            payment_info["payment_key"] = self.payment_key
+            result = self._request_with_params(
+                POST, MAKE_PAYMENT, self.orchestrator_url, payment_info, self.token
+            )
+            return result
+        except:
+            logging.error("An error occurred while executing make_payment()")
+            return None
+
+    def get_payment_info(self):
+        """Retrieve payment information from the orchestrator after making the payment."""
+        try:
+            payment_info = self._request_without_params(
+                GET, PAYMENT_INFO, self.orchestrator_url, self.token
+            )
+            return payment_info
+        except:
+            logging.error("An error occurred while executing get_payment_info()")
+            return None
+
+    def get_task_status(self):
+        """Fetch the current status of the task from the orchestrator."""
+        try:
+            task_status = self._request_without_params(
+                GET, TASK_STATUS, self.orchestrator_url, self.token
+            )
+            return task_status
+        except:
+            logging.error("An error occurred while executing get_task_status()")
+            return None
+
+    def fetch_task_details(self):
+        """Retrieve the deployed URL and credentials/token for access after task success.
+        Decrypt the credentials/token with the private key if necessary.
+        """
+        try:
+            task_details = self._request_without_params(
+                GET, TASK_DETAILS, self.orchestrator_url, self.token
+            )
+            return task_details
+        except:
+            logging.error("An error occurred while executing fetch_task_details()")
+            return None

swan/api_client.py

@@ -10,7 +10,7 @@
 
 class APIClient(object):
 
-    def __init__(self, api_key: str, login: bool = True,  environment: str = ""):
+    def __init__(self, api_key: str, login: bool = True, environment: str = ""):
         """Initialize user configuration and login
 
         Args:
@@ -31,15 +31,15 @@ def api_key_login(self):
             A str access token for further SwanHub API access in
             current session.
         """
-        params = {'api_key': self.api_key}
+        params = {"api_key": self.api_key}
         try:
             result = self._request_with_params(
                 POST, APIKEY_LOGIN, SWAN_API, params, None, None
             )
-            self.token = result['data']
-            logging.info('')
+            self.token = result["data"]
+            logging.info("")
         except:
-            logging.error('')
+            logging.error("")
 
     def _request(self, method, request_path, swan_api, params, token, files=False):
         if method == GET:
@@ -74,5 +74,7 @@ def _request(self, method, request_path, swan_api, params, token, files=False):
     def _request_without_params(self, method, request_path, swan_api, token):
         return self._request(method, request_path, swan_api, {}, token)
 
-    def _request_with_params(self, method, request_path, swan_api, params, token, files):
+    def _request_with_params(
+        self, method, request_path, swan_api, params, token, files
+    ):
         return self._request(method, request_path, swan_api, params, token, files)