Add support for Amazon States Language "ResultSelector" in Task, Map …

doc/placeholders.rst

@@ -1,12 +1,32 @@
 Placeholders
 =============
 
+
 Once defined, a workflow is static unless you update it explicitly. But, you can pass
 input to workflow executions. You can have dynamic values
-that you use in the **parameters** fields of the steps in your workflow. For this,
+that you use in the **parameters** or **result_selector** fields of the steps in your workflow. For this,
 the AWS Step Functions Data Science SDK provides a way to define placeholders to pass around when you
-create your workflow. There are 2 mechanisms for passing dynamic values in a workflow.
+create your workflow. There are 3 mechanisms for passing dynamic values in a workflow.
+
+.. autoclass:: stepfunctions.inputs.Placeholder
+
+.. autoclass:: stepfunctions.inputs.ExecutionInput
+                :inherited-members:
+
+.. autoclass:: stepfunctions.inputs.StepInput
+                :inherited-members:
+
+.. autoclass:: stepfunctions.inputs.StepResult
+                :inherited-members:
+
+- `Execution Input <#execution-input>`__
 
+- `Step Input <#step-input>`__
+
+- `Step Result <#step-result>`__
+
+Execution Input
+---------------
 The first mechanism is a global input to the workflow execution. This input is
 accessible to all the steps in the workflow. The SDK provides :py:meth:`stepfunctions.inputs.ExecutionInput`
 to define the schema for this input, and to access the values in your workflow.
@@ -50,6 +70,9 @@ to define the schema for this input, and to access the values in your workflow.
 
   workflow.execute(inputs={'myDynamicInput': "WorldHello"})
 
+
+Step Input
+----------
 The second mechanism is for passing dynamic values from one step to the next
 step. The output of one step becomes the input of the next step.
 The SDK provides the :py:meth:`stepfunctions.inputs.StepInput` class for this.
@@ -64,10 +87,10 @@ that returns the placeholder output for that step.
       parameters={
           "FunctionName": "MakeApiCall",
           "Payload": {
-            "input": "20192312"
-            }
+              "input": "20192312"
           }
-        )
+      }
+  )
 
   lambda_state_second = LambdaStep(
         state_id="MySecondLambdaStep",
@@ -82,11 +105,23 @@ that returns the placeholder output for that step.
   definition = Chain([lambda_state_first, lambda_state_second])
 
 
+Step Result
+-----------
+The third mechanism is a placeholder for a step's result. The result of a step can be modified
+with the **result_selector** field to replace the step's result.
 
-.. autoclass:: stepfunctions.inputs.Placeholder
+.. code-block:: python
 
-.. autoclass:: stepfunctions.inputs.ExecutionInput
-                :inherited-members:
+  lambda_result = StepResult(
+      schema={
+          "Id": str,
+      }
+  )
 
-.. autoclass:: stepfunctions.inputs.StepInput
-                :inherited-members:
+  lambda_state_first = LambdaStep(
+      state_id="MyFirstLambdaStep",
+      result_selector={
+          "Output": lambda_result["Id"],
+          "Status": "Success"
+      }
+  )

src/stepfunctions/inputs/__init__.py

@@ -6,10 +6,10 @@
 #
 #     http://www.apache.org/licenses/LICENSE-2.0
 #
-# or in the "license" file accompanying this file. This file 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 
+# or in the "license" file accompanying this file. This file 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.
 from __future__ import absolute_import
 
-from stepfunctions.inputs.placeholders import Placeholder, ExecutionInput, StepInput
+from stepfunctions.inputs.placeholders import Placeholder, ExecutionInput, StepInput, StepResult

src/stepfunctions/inputs/placeholders.py

@@ -6,9 +6,9 @@
 #
 #     http://www.apache.org/licenses/LICENSE-2.0
 #
-# or in the "license" file accompanying this file. This file 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 
+# or in the "license" file accompanying this file. This file 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.
 from __future__ import absolute_import
 
@@ -51,11 +51,11 @@ def __init__(self, schema=None, **kwargs):
             self._set_schema(schema)
             self._make_immutable()
         self.json_str_template = "{}"
-        
+
         self.name = kwargs.get("name")
         self.type = kwargs.get("type")
         self.parent = kwargs.get("parent")
-        
+
 
     def get(self, name, type):
         """
@@ -64,11 +64,11 @@ def get(self, name, type):
         Args:
             name (str): Name of the placeholder variable.
             type (type): Type of the placeholder variable.
-        
+
         Raises:
             ValueError: If placeholder variable with the same name but different type already exists.
             ValueError: If placeholder variable does not fit into a previously specified schema for the placeholder collection.
-        
+
         Returns:
             Placeholder: Placeholder variable.
         """
@@ -240,7 +240,7 @@ def _join_path(self, path):
     def to_jsonpath(self):
         """
         Returns a JSON path representation of the placeholder variable to be used for step parameters.
-        
+
         Returns:
             str: JSON path representation of the placeholder variable
         """
@@ -252,23 +252,25 @@ class ExecutionInput(Placeholder):
     """
         Top-level class for execution input placeholders.
     """
-    
+
     def __init__(self, schema=None, **kwargs):
         super(ExecutionInput, self).__init__(schema, **kwargs)
         self.json_str_template = '$$.Execution.Input{}'
 
     def _create_variable(self, name, parent, type=None):
         """
             Creates a placeholder variable for Workflow Input.
-            A placeholder variable can only be created if the collection is not immutable due to a pre-specified schema.
+            A placeholder variable can only be created if the collection is mutable.
+            A collection is mutable if no pre-specified schema was defined at construction.
         """
         if self.immutable:
-            raise ValueError("Placeholder variable does not conform to schema set for the placeholder collection.")
+            raise ValueError(f"Placeholder variable does not conform to schema set for the placeholder collection:"
+                             f" {self.schema}")
         if type:
             return ExecutionInput(name=name, parent=parent, type=type)
         else:
             return ExecutionInput(name=name, parent=parent)
-    
+
 
 class StepInput(Placeholder):
 
@@ -279,15 +281,42 @@ class StepInput(Placeholder):
     def __init__(self, schema=None, **kwargs):
         super(StepInput, self).__init__(schema, **kwargs)
         self.json_str_template = '${}'
-        
+
     def _create_variable(self, name, parent, type=None):
         """
             Creates a placeholder variable for Step Input.
-            A placeholder variable can only be created if the collection is not immutable due to a pre-specified schema.
+            A placeholder variable can only be created if the collection is mutable.
+            A collection is mutable if no pre-specified schema was defined at construction..
         """
         if self.immutable:
-            raise ValueError("Placeholder variable does not conform to schema set for the placeholder collection.")
+            raise ValueError(f"Placeholder variable does not conform to schema set for the placeholder collection:"
+                             f" {self.schema}")
         if type:
             return StepInput(name=name, parent=parent, type=type)
         else:
             return StepInput(name=name, parent=parent)
+
+
+class StepResult(Placeholder):
+
+    """
+        Top-level class for step result placeholders.
+    """
+
+    def __init__(self, schema=None, **kwargs):
+        super(StepResult, self).__init__(schema, **kwargs)
+        self.json_str_template = '${}'
+
+    def _create_variable(self, name, parent, type=None):
+        """
+            Creates a placeholder variable for Step Result.
+            A placeholder variable can only be created if the collection is mutable.
+            A collection is mutable if no pre-specified schema was defined at construction.
+        """
+        if self.immutable:
+            raise ValueError(f"Placeholder variable does not conform to schema set for the placeholder collection:"
+                             f" {self.schema}")
+        if type:
+            return StepResult(name=name, parent=parent, type=type)
+        else:
+            return StepResult(name=name, parent=parent)

src/stepfunctions/steps/compute.py

@@ -57,6 +57,7 @@ def __init__(self, state_id, wait_for_callback=False, **kwargs):
             comment (str, optional): Human-readable comment or description. (default: None)
             input_path (str, optional): Path applied to the state’s raw input to select some or all of it; that selection is used by the state. (default: '$')
             parameters (dict, optional): The value of this field becomes the effective input for the state.
+            result_selector (dict, optional): The value of this field becomes the effective result of the state.
             result_path (str, optional): Path specifying the raw input’s combination with or replacement by the state’s result. (default: '$')
             output_path (str, optional): Path applied to the state’s output after the application of `result_path`, producing the effective output which serves as the raw input for the next state. (default: '$')
         """
@@ -98,6 +99,7 @@ def __init__(self, state_id, wait_for_completion=True, **kwargs):
             comment (str, optional): Human-readable comment or description. (default: None)
             input_path (str, optional): Path applied to the state’s raw input to select some or all of it; that selection is used by the state. (default: '$')
             parameters (dict, optional): The value of this field becomes the effective input for the state.
+            result_selector (dict, optional): The value of this field becomes the effective result of the state.
             result_path (str, optional): Path specifying the raw input’s combination with or replacement by the state’s result. (default: '$')
             output_path (str, optional): Path applied to the state’s output after the application of `result_path`, producing the effective output which serves as the raw input for the next state. (default: '$')
         """
@@ -138,6 +140,7 @@ def __init__(self, state_id, wait_for_completion=True, **kwargs):
             comment (str, optional): Human-readable comment or description. (default: None)
             input_path (str, optional): Path applied to the state’s raw input to select some or all of it; that selection is used by the state. (default: '$')
             parameters (dict, optional): The value of this field becomes the effective input for the state.
+            result_selector (dict, optional): The value of this field becomes the effective result of the state.
             result_path (str, optional): Path specifying the raw input’s combination with or replacement by the state’s result. (default: '$')
             output_path (str, optional): Path applied to the state’s output after the application of `result_path`, producing the effective output which serves as the raw input for the next state. (default: '$')
         """
@@ -178,6 +181,7 @@ def __init__(self, state_id, wait_for_completion=True, **kwargs):
             comment (str, optional): Human-readable comment or description. (default: None)
             input_path (str, optional): Path applied to the state’s raw input to select some or all of it; that selection is used by the state. (default: '$')
             parameters (dict, optional): The value of this field becomes the effective input for the state.
+            result_selector (dict, optional): The value of this field becomes the effective result of the state.
             result_path (str, optional): Path specifying the raw input’s combination with or replacement by the state’s result. (default: '$')
             output_path (str, optional): Path applied to the state’s output after the application of `result_path`, producing the effective output which serves as the raw input for the next state. (default: '$')
         """

src/stepfunctions/steps/fields.py

@@ -22,6 +22,7 @@ class Field(Enum):
     InputPath        = 'input_path'
     OutputPath       = 'output_path'
     Parameters       = 'parameters'
+    ResultSelector   = 'result_selector'
     ResultPath       = 'result_path'
     Next             = 'next'
     Retry            = 'retry'