Pipeline CLI: Add support for static text spinner

elyra/cli/pipeline_app.py

@@ -20,15 +20,14 @@
 from operator import itemgetter
 import os
 from pathlib import Path
-from sys import exit
+import sys
 from typing import Optional
 import warnings
 
 import click
 from colorama import Fore
 from colorama import Style
 from kfp import Client as ArgoClient
-from yaspin import yaspin
 
 from elyra._version import __version__
 from elyra.metadata.manager import MetadataManager
@@ -48,6 +47,11 @@
 from elyra.pipeline.validation import PipelineValidationManager
 from elyra.pipeline.validation import ValidationSeverity
 
+if sys.stdout.isatty():
+    from yaspin import yaspin as Spinner
+else:
+    from .pipeline_app_utils import StaticTextSpinner as Spinner
+
 # custom exit code - a timeout occurred
 EXIT_TIMEDOUT = 124
 
@@ -237,7 +241,7 @@ def _execute_pipeline(pipeline_definition) -> PipelineProcessorResponse:
 
 def _build_component_cache():
     """Initialize a ComponentCache instance and wait for it to complete all tasks"""
-    with yaspin(text="Initializing the component cache..."):
+    with Spinner(text="Initializing the component cache..."):
         component_cache = ComponentCache.instance(emulate_server_app=True)
         component_cache.load()
         component_cache.wait_for_all_cache_tasks()
@@ -374,7 +378,7 @@ def submit(json_option, pipeline_path, runtime_config_name, monitor_option, time
     except Exception:
         raise click.ClickException("Pipeline validation FAILED. The pipeline was not submitted for execution.")
 
-    with yaspin(text="Submitting pipeline..."):
+    with Spinner(text="Submitting pipeline..."):
         response: PipelineProcessorResponse = _execute_pipeline(pipeline_definition)
 
     if not json_option:
@@ -406,22 +410,22 @@ def submit(json_option, pipeline_path, runtime_config_name, monitor_option, time
             msg = (
                 f"Monitoring status of pipeline run '{response.run_id}' for up to " f"{timeout_option} {minute_str}..."
             )
-            with yaspin(text=msg):
+            with Spinner(text=msg):
                 status = _monitor_kfp_submission(runtime_config, runtime_config_name, response.run_id, timeout_option)
         except TimeoutError:
             click.echo(
                 "Monitoring was stopped because the timeout threshold "
                 f"({timeout_option} {minute_str}) was exceeded. The pipeline is still running."
             )
-            exit(EXIT_TIMEDOUT)
+            sys.exit(EXIT_TIMEDOUT)
         else:
             # The following are known KFP states: 'succeeded', 'failed', 'skipped',
             # 'error'. Treat 'unknown' as error. Exit with appropriate status code.
             click.echo(f"Monitoring ended with run status: {status}")
             if status.lower() not in ["succeeded", "skipped"]:
                 # Click appears to use non-zero exit codes 1 (ClickException)
                 # and 2 (UsageError). Terminate.
-                exit(click.ClickException.exit_code)
+                sys.exit(click.ClickException.exit_code)
 
 
 def _monitor_kfp_submission(runtime_config: dict, runtime_config_name: str, run_id: str, timeout: int) -> str:
@@ -638,7 +642,7 @@ def export(pipeline_path, runtime_config, output, overwrite):
     except Exception:
         raise click.ClickException("Pipeline validation FAILED. The pipeline was not exported.")
 
-    with yaspin(text="Exporting pipeline ..."):
+    with Spinner(text="Exporting pipeline ..."):
         try:
             # parse pipeline
             pipeline_object = PipelineParser().parse(pipeline_definition)

elyra/cli/pipeline_app_utils.py

@@ -0,0 +1,40 @@
+#
+# Copyright 2018-2022 Elyra Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+
+class StaticTextSpinner:
+    """
+    A static text only spinner, implemented using the context manager interface.
+    https://docs.python.org/3/library/stdtypes.html#context-manager-types
+    """
+
+    def __init__(self, text: str):
+        self.enter_text = text
+
+    def __enter__(self):
+        """
+        Called by the with statement to enter the runtime context
+        https://docs.python.org/3/library/stdtypes.html#contextmanager.__enter__
+        """
+        print(self.enter_text)
+        return self
+
+    def __exit__(self, *args, **kwargs):
+        """
+        Called when the execution leaves the 'with' code block.
+        https://docs.python.org/3/library/stdtypes.html#contextmanager.__exit__
+        """
+        return False