feat: Add support for nested Aws StepFunctions service integration

doc/services.rst

@@ -20,6 +20,8 @@ This module provides classes to build steps that integrate with Amazon DynamoDB,
 
 - `Amazon SQS <#amazon-sqs>`__
 
+- `AWS Step Functions <#aws-step-functions>`__
+
 
 Amazon DynamoDB
 ----------------
@@ -82,3 +84,8 @@ Amazon SNS
 Amazon SQS
 -----------
 .. autoclass:: stepfunctions.steps.service.SqsSendMessageStep
+
+AWS Step Functions
+------------------
+.. autoclass:: stepfunctions.steps.service.StepFunctionsStartExecutionStep
+

src/stepfunctions/steps/__init__.py

@@ -34,3 +34,4 @@
 from stepfunctions.steps.service import EventBridgePutEventsStep
 from stepfunctions.steps.service import GlueDataBrewStartJobRunStep
 from stepfunctions.steps.service import SnsPublishStep, SqsSendMessageStep
+from stepfunctions.steps.service import StepFunctionsStartExecutionStep

src/stepfunctions/steps/integration_resources.py

@@ -25,6 +25,7 @@ class IntegrationPattern(Enum):
     WaitForTaskToken = "waitForTaskToken"
     WaitForCompletion = "sync"
     RequestResponse = ""
+    WaitForCompletionWithJsonResponse = "sync:2"
 
 
 def get_service_integration_arn(service, api, integration_pattern=IntegrationPattern.RequestResponse):

src/stepfunctions/steps/service.py

@@ -24,6 +24,7 @@
 GLUE_DATABREW_SERVICE_NAME = "databrew"
 SNS_SERVICE_NAME = "sns"
 SQS_SERVICE_NAME = "sqs"
+STEP_FUNCTIONS_SERVICE_NAME = "states"
 
 
 class DynamoDBApi(Enum):
@@ -70,6 +71,10 @@ class SqsApi(Enum):
     SendMessage = "sendMessage"
 
 
+class StepFunctions(Enum):
+    StartExecution = "startExecution"
+
+
 class DynamoDBGetItemStep(Task):
     """
     Creates a Task state to get an item from DynamoDB. See `Call DynamoDB APIs with Step Functions <https://docs.aws.amazon.com/step-functions/latest/dg/connect-ddb.html>`_ for more details.
@@ -887,3 +892,66 @@ def __init__(self, state_id, **kwargs):
                                                                    ElasticMapReduceApi.ModifyInstanceGroupByName)
 
         super(EmrModifyInstanceGroupByNameStep, self).__init__(state_id, **kwargs)
+
+
+class StepFunctionsStartExecutionStep(Task):
+
+    """
+    Creates a Task state that starts an execution of another state machine. See `Manage AWS Step Functions Executions as an Integrated Service <https://docs.aws.amazon.com/step-functions/latest/dg/connect-stepfunctions.html`_ for more details.
+
+    Property flags: There are three flags (wait_for_callback, wait_for_completion and async_call) that can be set in order to select which Step Functions resource to use.
+        One of three must be enabled to create the step successfully.
+    """
+
+    def __init__(self, state_id, wait_for_callback=False, wait_for_completion=True, async_call=False, **kwargs):
+        """
+        Args:
+            state_id (str): State name whose length **must be** less than or equal to 128 unicode characters. State names **must be** unique within the scope of the whole state machine.
+            wait_for_callback(bool, optional): Boolean value set to `True` if the Task state should wait for callback to resume the operation. (default: False)
+            wait_for_completion (bool, optional): Boolean value set to `True` if the Task state should wait to complete before proceeding to the next step in the workflow. (default: True)
+            async_call(bool, optional): Boolean value set to `True` for the Task proceed to the next step in the workflow without waiting for completion. (default: False):
+            timeout_seconds (int, optional): Positive integer specifying timeout for the state in seconds. If the state runs longer than the specified timeout, then the interpreter fails the state with a `States.Timeout` Error Name. (default: 60)
+            timeout_seconds_path (str, optional): Path specifying the state's timeout value in seconds from the state input. When resolved, the path must select a field whose value is a positive integer.
+            heartbeat_seconds (int, optional): Positive integer specifying heartbeat timeout for the state in seconds. This value should be lower than the one specified for `timeout_seconds`. If more time than the specified heartbeat elapses between heartbeats from the task, then the interpreter fails the state with a `States.Timeout` Error Name.
+            heartbeat_seconds_path (str, optional): Path specifying the state's heartbeat value in seconds from the state input. When resolved, the path must select a field whose value is a positive integer.
+            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_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: '$')
+        """
+        # Validate that only 1 one of the property flags are set to true
+        property_flags = [(wait_for_callback, "wait_for_callback"), (wait_for_completion, "wait_for_completion"),
+                          (async_call, "async_call")]
+        enabled_property_flags = [property_flag[1] for property_flag in property_flags if property_flag[0]]
+        if not enabled_property_flags:
+            raise ValueError(f"No resource flag enabled - Please enable one of "
+                             f"{[property_flag[1] for property_flag in property_flags]}")
+        elif len(enabled_property_flags) > 1:
+            raise ValueError(f"Multiple resource flags enabled({enabled_property_flags}) - "
+                             f"Please enable only one.")
+
+        if wait_for_callback:
+            """
+            Example resource arn: arn:aws:states:::states:startExecution.waitForTaskToken
+            """
+
+            kwargs[Field.Resource.value] = get_service_integration_arn(STEP_FUNCTIONS_SERVICE_NAME,
+                                                                       StepFunctions.StartExecution,
+                                                                       IntegrationPattern.WaitForTaskToken)
+        elif wait_for_completion:
+                """
+                Example resource arn:aws:states:::states:startExecution.sync:2
+                """
+
+                kwargs[Field.Resource.value] = get_service_integration_arn(STEP_FUNCTIONS_SERVICE_NAME,
+                                                                           StepFunctions.StartExecution,
+                                                                           IntegrationPattern.WaitForCompletionWithJsonResponse)
+        else:
+            """
+            Example resource arn:aws:states:::states:startExecution
+            """
+            kwargs[Field.Resource.value] = get_service_integration_arn(STEP_FUNCTIONS_SERVICE_NAME,
+                                                                       StepFunctions.StartExecution)
+
+        super(StepFunctionsStartExecutionStep, self).__init__(state_id, **kwargs)

tests/unit/test_service_steps.py

@@ -13,6 +13,7 @@
 from __future__ import absolute_import
 
 import boto3
+import pytest
 
 from unittest.mock import patch
 from stepfunctions.steps.service import DynamoDBGetItemStep, DynamoDBPutItemStep, DynamoDBUpdateItemStep, DynamoDBDeleteItemStep
@@ -30,6 +31,7 @@
 from stepfunctions.steps.service import EventBridgePutEventsStep
 from stepfunctions.steps.service import SnsPublishStep, SqsSendMessageStep
 from stepfunctions.steps.service import GlueDataBrewStartJobRunStep
+from stepfunctions.steps.service import StepFunctionsStartExecutionStep
 
 
 @patch.object(boto3.session.Session, 'region_name', 'us-east-1')
@@ -1158,3 +1160,102 @@ def test_eks_call_step_creation():
         },
         'End': True
     }
+
+
+@patch.object(boto3.session.Session, 'region_name', 'us-east-1')
+def test_step_functions_start_execution_step_creation():
+    step = StepFunctionsStartExecutionStep(
+        "SFN Start Execution", wait_for_callback=False, wait_for_completion=False, async_call=True,
+        parameters={
+            "Input": {
+                "Comment": "Hello world!"
+            },
+            "StateMachineArn": "arn:aws:states:us-east-1:123456789012:stateMachine:HelloWorld",
+            "Name": "ExecutionName"
+        })
+
+    assert step.to_dict() == {
+        "Type": "Task",
+        "Resource": "arn:aws:states:::states:startExecution",
+        "Parameters": {
+            "Input": {
+                "Comment": "Hello world!"
+            },
+            "StateMachineArn": "arn:aws:states:us-east-1:123456789012:stateMachine:HelloWorld",
+            "Name": "ExecutionName"
+        },
+        "End": True
+    }
+
+
+@patch.object(boto3.session.Session, 'region_name', 'us-east-1')
+def test_step_functions_start_execution_step_creation_sync():
+    step = StepFunctionsStartExecutionStep(
+        "SFN Start Execution - Sync", wait_for_callback=False, wait_for_completion=True, async_call=False,
+        parameters={
+            "Input": {
+                "Comment": "Hello world!"
+            },
+            "StateMachineArn": "arn:aws:states:us-east-1:123456789012:stateMachine:HelloWorld",
+            "Name": "ExecutionName"
+        })
+
+    assert step.to_dict() == {
+        "Type": "Task",
+        "Resource": "arn:aws:states:::states:startExecution.sync:2",
+        "Parameters": {
+            "Input": {
+                "Comment": "Hello world!"
+            },
+            "StateMachineArn": "arn:aws:states:us-east-1:123456789012:stateMachine:HelloWorld",
+            "Name": "ExecutionName"
+        },
+        "End": True
+    }
+
+
+@patch.object(boto3.session.Session, 'region_name', 'us-east-1')
+def test_step_functions_start_execution_step_creation_wait_for_callback():
+    step = StepFunctionsStartExecutionStep(
+        "SFN Start Execution - Wait for Callback", wait_for_callback=True, wait_for_completion=False, async_call=False,
+        parameters={
+            "Input": {
+                "Comment": "Hello world!",
+                "token.$": "$$.Task.Token"
+            },
+            "StateMachineArn": "arn:aws:states:us-east-1:123456789012:stateMachine:HelloWorld",
+            "Name": "ExecutionName"
+        })
+
+    assert step.to_dict() == {
+        "Type": "Task",
+        "Resource": "arn:aws:states:::states:startExecution.waitForTaskToken",
+        "Parameters": {
+            "Input": {
+                "Comment": "Hello world!",
+                "token.$": "$$.Task.Token"
+            },
+            "StateMachineArn": "arn:aws:states:us-east-1:123456789012:stateMachine:HelloWorld",
+            "Name": "ExecutionName"
+        },
+        "End": True
+    }
+
+
+@pytest.mark.parametrize("resource_flags", [
+    {'wait_for_callback': True, 'wait_for_completion': True, 'async_call': True},
+    {'wait_for_callback': False, 'wait_for_completion': True, 'async_call': True},
+    {'wait_for_callback': True, 'wait_for_completion': False, 'async_call': True},
+    {'wait_for_callback': True, 'wait_for_completion': True, 'async_call': False},
+])
+@patch.object(boto3.session.Session, 'region_name', 'us-east-1')
+def test_step_functions_start_execution_step_creation_multiple_enabled_flags_raises_exception(resource_flags):
+    with pytest.raises(ValueError):
+        StepFunctionsStartExecutionStep("SFN Start Execution - Multiple flags", **resource_flags)
+
+
+@patch.object(boto3.session.Session, 'region_name', 'us-east-1')
+def test_step_functions_start_execution_step_creation_no_enabled_flags_raises_exception():
+    with pytest.raises(ValueError):
+        StepFunctionsStartExecutionStep("SFN Start Execution - No active flags", wait_for_callback=False,
+                                        wait_for_completion=False, async_call=False)