change: Implement overriden solution for pipeline variables

Author: Dewen Qi

src/sagemaker/parameter.py

@@ -14,9 +14,6 @@
 from __future__ import absolute_import
 
 import json
-from sagemaker.workflow.parameters import Parameter as PipelineParameter
-from sagemaker.workflow.functions import JsonGet as PipelineJsonGet
-from sagemaker.workflow.functions import Join as PipelineJoin
 
 
 class ParameterRange(object):
@@ -72,12 +69,8 @@ def as_tuning_range(self, name):
         """
         return {
             "Name": name,
-            "MinValue": str(self.min_value)
-            if not isinstance(self.min_value, (PipelineParameter, PipelineJsonGet, PipelineJoin))
-            else self.min_value,
-            "MaxValue": str(self.max_value)
-            if not isinstance(self.max_value, (PipelineParameter, PipelineJsonGet, PipelineJoin))
-            else self.max_value,
+            "MinValue": str(self.min_value),
+            "MaxValue": str(self.max_value),
             "ScalingType": self.scaling_type,
         }
 
@@ -111,10 +104,7 @@ def __init__(self, values):  # pylint: disable=super-init-not-called
                 This input will be converted into a list of strings.
         """
         values = values if isinstance(values, list) else [values]
-        self.values = [
-            str(v) if not isinstance(v, (PipelineParameter, PipelineJsonGet, PipelineJoin)) else v
-            for v in values
-        ]
+        self.values = [str(v) for v in values]
 
     def as_tuning_range(self, name):
         """Represent the parameter range as a dictionary.

src/sagemaker/tuner.py

@@ -375,12 +375,7 @@ def _prepare_static_hyperparameters(
     ):
         """Prepare static hyperparameters for one estimator before tuning."""
         # Remove any hyperparameter that will be tuned
-        static_hyperparameters = {
-            str(k): str(v)
-            if not isinstance(v, (PipelineParameter, PipelineJsonGet, PipelineJoin))
-            else v
-            for (k, v) in estimator.hyperparameters().items()
-        }
+        static_hyperparameters = {str(k): str(v) for (k, v) in estimator.hyperparameters().items()}
         for hyperparameter_name in hyperparameter_ranges.keys():
             static_hyperparameters.pop(hyperparameter_name, None)
 

src/sagemaker/workflow/entities.py

@@ -14,9 +14,10 @@
 from __future__ import absolute_import
 
 import abc
+import json
 
 from enum import EnumMeta
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List, Union, Optional
 
 PrimitiveType = Union[str, int, bool, float, None]
 RequestType = Union[Dict[str, Any], List[Dict[str, Any]]]
@@ -57,3 +58,68 @@ class Expression(abc.ABC):
     @abc.abstractmethod
     def expr(self) -> RequestType:
         """Get the expression structure for workflow service calls."""
+
+
+class PipelineVariable(Expression):
+    """Base object for pipeline variables
+
+    PipelineVariables must implement the expr property.
+    """
+
+    def __add__(self, other: Union[Expression, PrimitiveType]):
+        """Add function for PipelineVariable
+
+        Args:
+            other (Union[Expression, PrimitiveType]): The other object to be added.
+
+        Return:
+            `sagemaker.workflow.functions.Join`
+        """
+        from sagemaker.workflow.functions import Join
+
+        if isinstance(self, Join):
+            self.values.append(other)  # pylint: disable=no-member
+            return self
+
+        return Join(on="", values=[self, other])
+
+    def __str__(self):
+        """String function for PipelineVariable"""
+        return json.dumps(self._expr_helper(True))
+
+    def _expr_helper(
+        self, cast_to_string: bool = False  # pylint: disable=unused-argument
+    ) -> RequestType:
+        """Get the expression structure for __str__ call.
+
+        Args:
+            cast_to_string (bool): To indicate if the expression value
+                is need to be casted to string in runtime.
+
+        Return:
+            Union[Dict[str, Any], List[Dict[str, Any]]]
+        """
+        return self.expr
+
+    @property
+    @abc.abstractmethod
+    def expr(self) -> RequestType:
+        """Get the expression structure for workflow service calls."""
+
+    def startswith(
+        self,
+        prefix: Union[str, tuple],  # pylint: disable=unused-argument
+        start: Optional[int] = None,  # pylint: disable=unused-argument
+        end: Optional[int] = None,  # pylint: disable=unused-argument
+    ) -> bool:
+        """Simulate the Python string's built-in method: startswith
+
+        Args:
+            prefix (str, tuple): The (tuple of) string to be checked.
+            start (int): To set the start index of the matching boundary (default: None).
+            end (int): To set the end index of the matching boundary (default: None).
+
+        Return:
+            bool: Always return False as Pipeline variables are parsed during execution runtime
+        """
+        return False

src/sagemaker/workflow/execution_variables.py

@@ -13,13 +13,15 @@
 """Pipeline parameters and conditions for workflow."""
 from __future__ import absolute_import
 
+import json
+
 from sagemaker.workflow.entities import (
-    Expression,
     RequestType,
+    PipelineVariable,
 )
 
 
-class ExecutionVariable(Expression):
+class ExecutionVariable(PipelineVariable):
     """Pipeline execution variables for workflow."""
 
     def __init__(self, name: str):
@@ -30,6 +32,10 @@ def __init__(self, name: str):
         """
         self.name = name
 
+    def __str__(self):
+        """String function for ExecutionVariable"""
+        return json.dumps(self.expr)
+
     @property
     def expr(self) -> RequestType:
         """The 'Get' expression dict for an `ExecutionVariable`."""

src/sagemaker/workflow/functions.py

@@ -13,16 +13,17 @@
 """The step definitions for workflow."""
 from __future__ import absolute_import
 
+import json
 from typing import List, Union
 
 import attr
 
-from sagemaker.workflow.entities import Expression
+from sagemaker.workflow.entities import PipelineVariable
 from sagemaker.workflow.properties import PropertyFile
 
 
 @attr.s
-class Join(Expression):
+class Join(PipelineVariable):
     """Join together properties.
 
     Examples:
@@ -38,27 +39,39 @@ class Join(Expression):
     Attributes:
         values (List[Union[PrimitiveType, Parameter, Expression]]):
             The primitive type values, parameters, step properties, expressions to join.
-        on_str (str): The string to join the values on (Defaults to "").
+        on (str): The string to join the values on (Defaults to "").
     """
 
     on: str = attr.ib(factory=str)
     values: List = attr.ib(factory=list)
 
+    def __str__(self):
+        """String function for Join"""
+        return json.dumps(self.expr)
+
     @property
     def expr(self):
         """The expression dict for a `Join` function."""
+        from sagemaker.workflow.utilities import is_pipeline_variable_expr_string
+
+        values = list()
+        for value in self.values:
+            if hasattr(value, "expr"):
+                values.append(value.expr)
+            elif is_pipeline_variable_expr_string(value):
+                values.append(json.loads(value))
+            else:
+                values.append(value)
         return {
             "Std:Join": {
                 "On": self.on,
-                "Values": [
-                    value.expr if hasattr(value, "expr") else value for value in self.values
-                ],
+                "Values": values,
             },
         }
 
 
 @attr.s
-class JsonGet(Expression):
+class JsonGet(PipelineVariable):
     """Get JSON properties from PropertyFiles.
 
     Attributes:
@@ -72,6 +85,21 @@ class JsonGet(Expression):
     property_file: Union[PropertyFile, str] = attr.ib()
     json_path: str = attr.ib()
 
+    def _expr_helper(self, cast_to_string: bool = False):
+        """Get the expression structure for __str__ call.
+
+        Args:
+            cast_to_string (bool): To indicate if the expression value
+                is need to be casted to string in runtime.
+
+        Return:
+            Union[Dict[str, Any], List[Dict[str, Any]]]
+        """
+        expression = self.expr
+        if cast_to_string:
+            expression["Std:JsonGet"]["CastToString"] = cast_to_string
+        return expression
+
     @property
     def expr(self):
         """The expression dict for a `JsonGet` function."""
@@ -82,6 +110,7 @@ def expr(self):
             name = self.property_file.name
         else:
             name = self.property_file
+
         return {
             "Std:JsonGet": {
                 "PropertyFile": {"Get": f"Steps.{self.step_name}.PropertyFiles.{name}"},

src/sagemaker/workflow/parameters.py

@@ -13,6 +13,7 @@
 """Pipeline parameters and conditions for workflow."""
 from __future__ import absolute_import
 
+import json
 from enum import Enum
 from functools import partial
 from typing import Dict, List, Type
@@ -24,6 +25,7 @@
     Entity,
     PrimitiveType,
     RequestType,
+    PipelineVariable,
 )
 
 
@@ -48,7 +50,7 @@ def python_type(self) -> Type:
 
 
 @attr.s
-class Parameter(Entity):
+class Parameter(PipelineVariable, Entity):
     """Pipeline parameter for workflow.
 
     Attributes:
@@ -84,6 +86,21 @@ def to_request(self) -> RequestType:
             value["DefaultValue"] = self.default_value
         return value
 
+    def _expr_helper(self, cast_to_string: bool = False) -> RequestType:
+        """Get the expression structure for __str__ call.
+
+        Args:
+            cast_to_string (bool): To indicate if the expression value
+                is need to be casted to string in runtime.
+
+        Return:
+            Union[Dict[str, Any], List[Dict[str, Any]]]
+        """
+        expression = self.expr
+        if cast_to_string:
+            expression["Get"] += "._CastToString"
+        return expression
+
     @property
     def expr(self) -> Dict[str, str]:
         """The 'Get' expression dict for a `Parameter`."""
@@ -170,6 +187,10 @@ def __hash__(self):
         """Hash function for parameter types"""
         return hash(tuple(self.to_request()))
 
+    def __str__(self):
+        """String function for ParameterString"""
+        return json.dumps(self.expr)
+
     def to_request(self) -> RequestType:
         """Get the request structure for workflow service calls."""
         request_dict = super(ParameterString, self).to_request()

src/sagemaker/workflow/pipeline.py

@@ -39,7 +39,7 @@
 from sagemaker.workflow.properties import Properties
 from sagemaker.workflow.steps import Step
 from sagemaker.workflow.step_collections import StepCollection
-from sagemaker.workflow.utilities import list_to_request
+from sagemaker.workflow.utilities import list_to_request, is_pipeline_variable_expr_string
 
 
 @attr.s
@@ -309,7 +309,6 @@ def definition(self) -> str:
             callback_output_to_step_map=callback_output_to_step_map,
             lambda_output_to_step_map=lambda_output_to_step_name,
         )
-
         return json.dumps(request_dict)
 
 
@@ -380,6 +379,8 @@ def _interpolate(
             interpolate(value, callback_output_to_step_map, lambda_output_to_step_map)
             for value in obj
         )
+    elif is_pipeline_variable_expr_string(obj):
+        return json.loads(obj)
     else:
         return obj
     return new

src/sagemaker/workflow/properties.py

@@ -13,16 +13,17 @@
 """The properties definitions for workflow."""
 from __future__ import absolute_import
 
+from abc import ABCMeta
 from typing import Dict, Union, List
 
 import attr
 
 import botocore.loaders
 
-from sagemaker.workflow.entities import Expression
+from sagemaker.workflow.entities import Expression, PipelineVariable, RequestType
 
 
-class PropertiesMeta(type):
+class PropertiesMeta(ABCMeta):
     """Load an internal shapes attribute from the botocore service model
 
     for sagemaker and emr service.
@@ -44,7 +45,7 @@ def __new__(mcs, *args, **kwargs):
         return super().__new__(mcs, *args, **kwargs)
 
 
-class Properties(metaclass=PropertiesMeta):
+class Properties(PipelineVariable, metaclass=PropertiesMeta):
     """Properties for use in workflow expressions."""
 
     def __init__(
@@ -61,6 +62,7 @@ def __init__(
             shape_name (str): The botocore service model shape name.
             shape_names (str): A List of the botocore service model shape name.
         """
+        # super().__init__()
         self._path = path
         shape_names = [] if shape_names is None else shape_names
         self._shape_names = shape_names if shape_name is None else [shape_name] + shape_names
@@ -88,6 +90,21 @@ def __init__(
                             f"{path}.{key}", info["shape"], service_name=service_name
                         )
 
+    def _expr_helper(self, cast_to_string: bool = False) -> RequestType:
+        """Get the expression structure for __str__ call.
+
+        Args:
+            cast_to_string (bool): To indicate if the expression value
+                is need to be casted to string in runtime.
+
+        Return:
+            Union[Dict[str, Any], List[Dict[str, Any]]]
+        """
+        expression = self.expr
+        if cast_to_string:
+            expression["Get"] += "._CastToString"
+        return expression
+
     @property
     def expr(self):
         """The 'Get' expression dict for a `Properties`."""