Load processor only if library is found

Author: akchinSTC

elyra/metadata/schemasproviders.py

@@ -29,6 +29,20 @@
     # We may not have kfp-tekton available and that's okay!
     TektonClient = None
 
+try:
+    from kfp import compiler  # noqa
+except ImportError:
+    # KFP python package is not installed
+    compiler = None
+
+try:
+    # Check to see if Airflow package is installed, since we do not have any dependencies
+    # on any Airflow package, we use GitHub as our canary
+    from github import Github  # noqa
+except ImportError:
+    # Github package is not installed, we use GitHub as our default DAG repository
+    Github = None
+
 from elyra.metadata.schema import SchemasProvider
 from elyra.metadata.schemaspaces import CodeSnippets
 from elyra.metadata.schemaspaces import ComponentCatalogs
@@ -51,7 +65,7 @@ class ElyraSchemasProvider(SchemasProvider, metaclass=ABCMeta):
             schema_json = json.load(f)
             local_schemas.append(schema_json)
 
-    def __init__(self):
+    def __init__(self, *args, **kwargs):
         self.log = log.get_logger()
         # get set of registered runtimes
         self._runtime_processor_names = set()
@@ -82,17 +96,17 @@ def get_schemas(self) -> List[Dict]:
         schemas = self.get_local_schemas_by_schemaspace(Runtimes.RUNTIMES_SCHEMASPACE_ID)
         for schema in schemas:
             if schema["name"] in self._runtime_processor_names:
-                runtime_schemas.append(schema)
-                if schema["name"] == "kfp":
+                if schema["name"] == "kfp" and compiler:
                     kfp_schema_present = True
-                elif schema["name"] == "airflow":
+                    runtime_schemas.append(schema)
+                elif schema["name"] == "airflow" and Github:
                     airflow_schema_present = True
+                    runtime_schemas.append(schema)
             else:
                 self.log.error(
                     f"No entrypoint with name '{schema['name']}' was found in group "
                     f"'elyra.pipeline.processor' to match the schema with the same name. Skipping..."
                 )
-
         if kfp_schema_present:  # Update the kfp engine enum to reflect current packages...
             # If TektonClient package is missing, navigate to the engine property
             # and remove 'tekton' entry if present and return updated result.

elyra/pipeline/processor.py

@@ -35,6 +35,8 @@
 from urllib3.exceptions import MaxRetryError
 
 from elyra.metadata.manager import MetadataManager
+from elyra.metadata.schema import SchemaManager
+from elyra.metadata.schemaspaces import Runtimes
 from elyra.pipeline.component import Component
 from elyra.pipeline.component_catalog import ComponentCache
 from elyra.pipeline.pipeline import GenericOperation
@@ -57,19 +59,23 @@ def __init__(self, **kwargs):
         self.root_dir = get_expanded_path(kwargs.get("root_dir"))
         # Register all known processors based on entrypoint configuration
         for processor in entrypoints.get_group_all("elyra.pipeline.processors"):
-            try:
-                # instantiate an actual instance of the processor
-                processor_instance = processor.load()(self.root_dir, parent=kwargs.get("parent"))  # Load an instance
-                self.log.info(
-                    f"Registering {processor.name} processor " f'"{processor.module_name}.{processor.object_name}"...'
-                )
-                self.add_processor(processor_instance)
-            except Exception as err:
-                # log and ignore initialization errors
-                self.log.error(
-                    f"Error registering {processor.name} processor "
-                    f'"{processor.module_name}.{processor.object_name}" - {err}'
-                )
+            if processor.name in SchemaManager.instance().get_schemasproviders(Runtimes.RUNTIMES_SCHEMASPACE_ID):
+                try:
+                    # instantiate an actual instance of the processor
+                    processor_instance = processor.load()(
+                        self.root_dir, parent=kwargs.get("parent")
+                    )  # Load an instance
+                    self.log.info(
+                        f"Registering {processor.name} processor "
+                        f'"{processor.module_name}.{processor.object_name}"...'
+                    )
+                    self.add_processor(processor_instance)
+                except Exception as err:
+                    # log and ignore initialization errors
+                    self.log.error(
+                        f"Error registering {processor.name} processor "
+                        f'"{processor.module_name}.{processor.object_name}" - {err}'
+                    )
 
     def add_processor(self, processor):
         self.log.debug(f"Registering {processor.type.value} runtime processor '{processor.name}'")

setup.cfg

@@ -155,9 +155,7 @@ papermill.engine =
 [options.packages.find]
 exclude =
     docs*
-    tests
-    tests.*
-    elyra.tests.*
+    tests*
 
 [flake8]
 application-import-names = elyra, test