[ML][Pipelines] refactor for reference from other packages (Azure#31929)

* fix: update message when user use an internal spark when not enabled

* refactor: divide validation related code for further vendor

* fix: support load run.yaml of a flow with additional_includes

Author: elliotzh

sdk/ml/azure-ai-ml/azure/ai/ml/_internal/_setup.py

@@ -1,11 +1,13 @@
 # ---------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
+from typing import NoReturn
 
 # pylint: disable=protected-access
 from marshmallow import INCLUDE
 
 from .._schema import NestedField
+from ..entities._builders.control_flow_node import LoopNode
 from ..entities._component.component_factory import component_factory
 from ..entities._job.pipeline._load_component import pipeline_node_factory
 from ._schema.command import CommandSchema, DistributedSchema, ParallelSchema
@@ -20,6 +22,7 @@
     InternalBaseNode,
     InternalComponent,
     Parallel,
+    Pipeline,
     Scope,
     Starlite,
 )
@@ -57,7 +60,14 @@ def _register_node(_type, node_cls, schema_cls):
     )
 
 
-def enable_internal_components_in_pipeline(*, force=False):
+def enable_internal_components_in_pipeline(*, force=False) -> NoReturn:
+    """Enable internal components in pipeline.
+
+    :keyword force: Whether to force re-enable internal components. Defaults to False.
+    :type force: bool
+    :return: No return value.
+    :rtype: None
+    """
     if _registered and not force:
         return  # already registered
 
@@ -83,4 +93,8 @@ def enable_internal_components_in_pipeline(*, force=False):
     _register_node(NodeType.SCOPE_V2, Scope, ScopeSchema)
     _register_node(NodeType.HDI_V2, HDInsight, HDInsightSchema)
     # Ae365exepool and AetherBridge have been registered to InternalBaseNode
+
+    # allow using internal nodes in do-while loop
+    LoopNode._extra_body_types = (Command, Pipeline)
+
     _set_registered(True)

sdk/ml/azure-ai-ml/azure/ai/ml/_internal/entities/environment.py

@@ -7,8 +7,8 @@
 from typing import Dict, Optional, Union
 
 from ..._utils.utils import load_yaml
-from ...constants._common import DefaultOpenEncoding, FILE_PREFIX
-from ...entities._validation import MutableValidationResult, _ValidationResultBuilder
+from ...constants._common import FILE_PREFIX, DefaultOpenEncoding
+from ...entities._validation import MutableValidationResult, ValidationResultBuilder
 
 
 class InternalEnvironment:
@@ -45,7 +45,7 @@ def _parse_file_path(value: str) -> str:
     def _validate_conda_section(
         self, base_path: Union[str, PathLike], skip_path_validation: bool
     ) -> MutableValidationResult:
-        validation_result = _ValidationResultBuilder.success()
+        validation_result = ValidationResultBuilder.success()
         if not self.conda:
             return validation_result
         dependencies_field_names = {self.CONDA_DEPENDENCIES, self.CONDA_DEPENDENCIES_FILE, self.PIP_REQUIREMENTS_FILE}
@@ -74,7 +74,7 @@ def _validate_conda_section(
     def _validate_docker_section(
         self, base_path: Union[str, PathLike], skip_path_validation: bool
     ) -> MutableValidationResult:
-        validation_result = _ValidationResultBuilder.success()
+        validation_result = ValidationResultBuilder.success()
         if not self.docker:
             return validation_result
         if not self.docker.get(self.BUILD) or not self.docker[self.BUILD].get(self.DOCKERFILE):
@@ -104,7 +104,7 @@ def validate(self, base_path: Union[str, PathLike], skip_path_validation: bool =
         :return: The validation result
         :rtype: MutableValidationResult
         """
-        validation_result = _ValidationResultBuilder.success()
+        validation_result = ValidationResultBuilder.success()
         if self.os is not None and self.os not in {"Linux", "Windows", "linux", "windows"}:
             validation_result.append_error(
                 yaml_path="os",

sdk/ml/azure-ai-ml/azure/ai/ml/_internal/entities/spark.py

@@ -17,7 +17,7 @@
 
 class InternalSparkComponent(
     InternalComponent, ParameterizedSpark, SparkJobEntryMixin
-):  # pylint: disable=too-many-instance-attributes
+):  # pylint: disable=too-many-instance-attributes, too-many-ancestors
     """Internal Spark Component
     This class is used to handle internal spark component.
     It can be loaded from internal spark component yaml or from rest object of an internal spark component.
@@ -91,13 +91,25 @@ def _create_schema_for_validation(cls, context) -> Union[PathAwareSchema, Schema
         return InternalSparkComponentSchema(context=context)
 
     @property
-    def environment(self):
+    def environment(self) -> Union[str, Environment, dict]:
+        """Get the environment of the component.
+
+        :return: The environment of the component.
+        :rtype: Union[str, Environment, dict]
+        """
         if isinstance(self._environment, Environment) and self._environment.image is None:
             return Environment(conda_file=self._environment.conda_file, image=DUMMY_IMAGE)
         return self._environment
 
     @environment.setter
     def environment(self, value):
+        """Set the environment of the component.
+
+        :param value: The environment of the component.
+        :type value: Union[str, Environment, dict]
+        :return: No return
+        :rtype: None
+        """
         if value is None or isinstance(value, (str, Environment)):
             self._environment = value
         elif isinstance(value, dict):
@@ -119,21 +131,45 @@ def environment(self, value):
             raise ValueError(f"Unsupported environment type: {type(value)}")
 
     @property
-    def jars(self):
+    def jars(self) -> List[str]:
+        """Get the jars of the component.
+
+        :return: The jars of the component.
+        :rtype: List[str]
+        """
         return self._jars
 
     @jars.setter
-    def jars(self, value):
+    def jars(self, value: Union[str, List[str]]):
+        """Set the jars of the component.
+
+        :param value: The jars of the component.
+        :type value: Union[str, List[str]]
+        :return: No return
+        :rtype: None
+        """
         if isinstance(value, str):
             value = [value]
         self._jars = value
 
     @property
-    def py_files(self):
+    def py_files(self) -> List[str]:
+        """Get the py_files of the component.
+
+        :return: The py_files of the component.
+        :rtype: List[str]
+        """
         return self._py_files
 
     @py_files.setter
     def py_files(self, value):
+        """Set the py_files of the component.
+
+        :param value: The py_files of the component.
+        :type value: Union[str, List[str]]
+        :return: No return
+        :rtype: None
+        """
         if isinstance(value, str):
             value = [value]
         self._py_files = value