Add support for Amazon States Language "ResultSelector" in Task, Map and Parallel States.

Author: Daniel Yoo

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,7 +252,7 @@ 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{}'
@@ -268,7 +268,7 @@ def _create_variable(self, name, parent, type=None):
             return ExecutionInput(name=name, parent=parent, type=type)
         else:
             return ExecutionInput(name=name, parent=parent)
-    
+
 
 class StepInput(Placeholder):
 
@@ -279,7 +279,7 @@ 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.
@@ -291,3 +291,26 @@ def _create_variable(self, name, parent, type=None):
             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 not immutable due to a pre-specified schema.
+        """
+        if self.immutable:
+            raise ValueError("Placeholder variable does not conform to schema set for the placeholder collection.")
+        if type:
+            return StepResult(name=name, parent=parent, type=type)
+        else:
+            return StepResult(name=name, parent=parent)

src/stepfunctions/steps/compute.py

@@ -32,6 +32,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: '$')
         """
@@ -59,6 +60,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: '$')
         """
@@ -86,6 +88,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: '$')
         """
@@ -113,6 +116,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

@@ -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
 
@@ -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'