How to use the stepfunctions.steps.Task function in stepfunctions

To help you get started, we’ve selected a few stepfunctions examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github aws / aws-step-functions-data-science-sdk-python / tests / unit / test_steps.py View on Github external
def test_paths_none_converted_to_null():
    task_state = Task('Task', resource='arn:aws:lambda:us-east-1:1234567890:function:StartLambda',
                      result_path=None,
                      input_path=None,
                      output_path=None)
    assert '"ResultPath": null' in task_state.to_json()
    assert '"InputPath": null' in task_state.to_json()
    assert '"OutputPath": null' in task_state.to_json()
github aws / aws-step-functions-data-science-sdk-python / tests / unit / test_steps.py View on Github external
def test_default_paths_not_included():
    task_state = Task('Task', resource='arn:aws:lambda:us-east-1:1234567890:function:StartLambda')
    assert 'ResultPath' not in task_state.to_dict()
    assert 'InputPath' not in task_state.to_dict()
    assert 'OutputPath' not in task_state.to_dict()
github aws / aws-step-functions-data-science-sdk-python / tests / unit / test_placeholders_with_steps.py View on Github external
def test_choice_state_with_placeholders():

    first_state = Task('FirstState', resource='arn:aws:lambda:us-east-1:1234567890:function:FirstState')
    retry = Chain([Pass('Retry'), Pass('Cleanup'), first_state])

    choice_state = Choice('Is Completed?')
    choice_state.add_choice(
        ChoiceRule.BooleanEquals(choice_state.output()["Completed"], True), 
        Succeed('Complete')
    )
    choice_state.add_choice(
        ChoiceRule.BooleanEquals(choice_state.output()["Completed"], False), 
        retry
    )

    first_state.next(choice_state)

    result = Graph(first_state).to_dict()
github aws / aws-step-functions-data-science-sdk-python / tests / unit / test_steps.py View on Github external
def test_paths_none():
    task_state = Task('Task', resource='arn:aws:lambda:us-east-1:1234567890:function:StartLambda',
                      result_path=None,
                      input_path=None,
                      output_path=None)
    assert 'ResultPath' in task_state.to_dict()
    assert task_state.to_dict()['ResultPath'] is None

    assert 'InputPath' in task_state.to_dict()
    assert task_state.to_dict()['InputPath'] is None

    assert 'OutputPath' in task_state.to_dict()
    assert task_state.to_dict()['OutputPath'] is None
github aws / aws-step-functions-data-science-sdk-python / tests / unit / test_steps.py View on Github external
def test_task_state_creation():
    task_state = Task('Task', resource='arn:aws:lambda:us-east-1:1234567890:function:StartLambda')
    task_state.add_retry(Retry(error_equals=['ErrorA', 'ErrorB'], interval_seconds=1, max_attempts=2, backoff_rate=2))
    task_state.add_retry(Retry(error_equals=['ErrorC'], interval_seconds=5))
    task_state.add_catch(Catch(error_equals=['States.ALL'], next_step=Pass('End State')))
    assert task_state.type == 'Task'
    assert len(task_state.retries) == 2
    assert len(task_state.catches) == 1
    assert task_state.to_dict() == {
        'Type': 'Task',
        'Resource': 'arn:aws:lambda:us-east-1:1234567890:function:StartLambda',
        'Retry': [
            {
                'ErrorEquals': ['ErrorA', 'ErrorB'],
                'IntervalSeconds': 1,
                'BackoffRate': 2,
                'MaxAttempts': 2
            },
github aws / aws-step-functions-data-science-sdk-python / tests / unit / test_steps.py View on Github external
def test_default_paths_not_converted_to_null():
    task_state = Task('Task', resource='arn:aws:lambda:us-east-1:1234567890:function:StartLambda')
    assert '"ResultPath": null' not in task_state.to_json()
    assert '"InputPath": null' not in task_state.to_json()
    assert '"OutputPath": null' not in task_state.to_json()