adding Pipelines to the Artifacts View and REPL

src/sageworks/api/meta.py

@@ -14,6 +14,7 @@
 from sageworks.utils.config_manager import ConfigManager
 from sageworks.utils.datetime_utils import datetime_string
 from sageworks.utils.aws_utils import num_columns_ds, num_columns_fs, aws_url
+from sageworks.api.pipeline_manager import PipelineManager
 
 
 class Meta:
@@ -37,6 +38,9 @@ def __init__(self):
         self.aws_broker = AWSServiceBroker()
         self.cm = ConfigManager()
 
+        # Pipeline Manager
+        self.pipeline_manager = PipelineManager()
+
     def account(self) -> dict:
         """Print out the AWS Account Info
 
@@ -344,6 +348,20 @@ def endpoints_deep(self, refresh: bool = False) -> dict:
         """
         return self.aws_broker.get_metadata(ServiceCategory.ENDPOINTS, force_refresh=refresh)
 
+    def pipelines(self, refresh: bool = False) -> pd.DataFrame:
+        """Get a summary of the SageWorks Pipelines
+
+        Args:
+            refresh (bool, optional): Force a refresh of the metadata. Defaults to False.
+
+        Returns:
+            pd.DataFrame: A summary of the SageWorks Pipelines
+        """
+        data = self.pipeline_manager.list_pipelines()
+
+        # Return the pipelines summary as a DataFrame
+        return pd.DataFrame(data)
+
     def _remove_sageworks_meta(self, data: dict) -> dict:
         """Internal: Recursively remove any keys with 'sageworks_' in them"""
 
@@ -400,6 +418,10 @@ def _remove_sageworks_meta(self, data: dict) -> dict:
     print("\n\n*** Endpoints ***")
     pprint(meta.endpoints())
 
+    # Get the Pipelines
+    print("\n\n*** Pipelines ***")
+    pprint(meta.pipelines())
+
     # Now do a deep dive on all the Artifacts
     print("\n\n#")
     print("# Deep Dives ***")

src/sageworks/api/pipeline.py

@@ -105,6 +105,17 @@ def _get_pipeline(self) -> dict:
         json_object = json.loads(response["Body"].read())
         return json_object
 
+    def __repr__(self) -> str:
+        """String representation of this pipeline
+
+        Returns:
+            str: String representation of this pipeline
+        """
+        # Class name and details
+        class_name = self.__class__.__name__
+        details = json.dumps(self.details(), indent=4)
+        return f"{class_name}({details})"
+
 
 if __name__ == "__main__":
     """Exercise the Pipeline Class"""
@@ -116,3 +127,6 @@ def _get_pipeline(self) -> dict:
 
     # Execute the Pipeline
     my_pipeline.execute()
+
+    # Print the Representation of the Pipeline
+    print(my_pipeline)

src/sageworks/repl/sageworks_shell.py

@@ -96,6 +96,7 @@ def __init__(self):
         self.commands["feature_sets"] = self.feature_sets
         self.commands["models"] = self.models
         self.commands["endpoints"] = self.endpoints
+        self.commands["pipelines"] = self.pipelines
         self.commands["log_debug"] = self.log_debug
         self.commands["log_info"] = self.log_info
         self.commands["log_important"] = self.log_important
@@ -209,6 +210,8 @@ def import_sageworks(self):
         self.commands["Endpoint"] = importlib.import_module("sageworks.api.endpoint").Endpoint
         self.commands["Monitor"] = importlib.import_module("sageworks.api.monitor").Monitor
         self.commands["Meta"] = importlib.import_module("sageworks.api.meta").Meta
+        self.commands["Pipeline"] = importlib.import_module("sageworks.api.pipeline").Pipeline
+        self.commands["PipelineManager"] = importlib.import_module("sageworks.api.pipeline_manager").PipelineManager
         self.commands["PluginManager"] = importlib.import_module("sageworks.utils.plugin_manager").PluginManager
 
         # These are 'nice to have' imports
@@ -241,6 +244,7 @@ def help_txt():
         - feature_sets: List all the FeatureSets in AWS
         - models: List all the Models in AWS
         - endpoints: List all the Endpoints in AWS
+        - pipelines: List all the SageWorks Pipelines
         - config: Show the current SageWorks Config
         - status: Show the current SageWorks Status
         - log_(debug/info/important/warning): Set the SageWorks log level
@@ -266,10 +270,15 @@ def summary(self):
             # Pad the name to 15 characters
             name = (name + " " * 15)[:15]
 
+            # Sanity check the dataframe
+            if df.empty:
+                examples = ""
+
             # Get the first three items in the first column
-            examples = ", ".join(df.iloc[:, 0].tolist())
-            if len(examples) > 70:
-                examples = examples[:70] + "..."
+            else:
+                examples = ", ".join(df.iloc[:, 0].tolist())
+                if len(examples) > 70:
+                    examples = examples[:70] + "..."
 
             # Print the summary
             cprint(["lightpurple", "\t" + name, "lightgreen", str(df.shape[0]) + "  ", "purple_blue", examples])
@@ -292,6 +301,9 @@ def models(self):
     def endpoints(self):
         return self.artifacts_text_view.endpoints_summary()
 
+    def pipelines(self):
+        return self.artifacts_text_view.pipelines_summary()
+
     @staticmethod
     def log_debug():
         logging.getLogger("sageworks").setLevel(logging.DEBUG)

src/sageworks/views/artifacts_text_view.py

@@ -41,6 +41,7 @@ def view_data(self) -> Dict[str, pd.DataFrame]:
             "FEATURE_SETS": self.feature_sets_summary(),
             "MODELS": self.models_summary(),
             "ENDPOINTS": self.endpoints_summary(),
+            "PIPELINES": self.pipelines_summary(),
         }
         return summary_data
 
@@ -79,6 +80,10 @@ def endpoints_summary(self) -> pd.DataFrame:
         """Get summary data about the SageWorks Endpoints"""
         return self.meta.endpoints()
 
+    def pipelines_summary(self) -> pd.DataFrame:
+        """Get summary data about the SageWorks Pipelines"""
+        return self.meta.pipelines()
+
 
 if __name__ == "__main__":
     import time