awsstepfuncs package

AWS Step Functions.

Submodules

awsstepfuncs.abstract_state module

Abstract state definitions.

Based on this table: https://states-language.net/spec.html#state-type-table

class awsstepfuncs.abstract_state.AbstractInputPathOutputPathState(*args, input_path='$', output_path='$', **kwargs)

Bases: awsstepfuncs.abstract_state.AbstractState

An Amazon States Language state including InputPath and OutputPath.

input_path and output_path let you control what is input and output from a state by using Reference Paths.

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

simulate(state_input, resource_to_mock_fn)

Simulate the state including input and output processing.

Parameters
  • state_input (Any) – The input to the state.

  • resource_to_mock_fn (Dict[str, Callable]) – A mapping of resource URIs to mock functions to use if the state performs a task.

Returns

The output of the state after applying any output processing.

Return type

Any

class awsstepfuncs.abstract_state.AbstractNextOrEndState(*args, input_path='$', output_path='$', **kwargs)

Bases: awsstepfuncs.abstract_state.AbstractInputPathOutputPathState

An Amazon States Language state including Next or End.

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

class awsstepfuncs.abstract_state.AbstractParametersState(*args, parameters=None, **kwargs)

Bases: awsstepfuncs.abstract_state.AbstractResultPathState

An Amazon States Language state including Parameters.

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

class awsstepfuncs.abstract_state.AbstractResultPathState(*args, result_path='$', **kwargs)

Bases: awsstepfuncs.abstract_state.AbstractNextOrEndState

An Amazon States Language state including ResultPath.

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

simulate(state_input, resource_to_mock_fn)

Simulate the state including input and output processing.

Parameters
  • state_input (Any) – The input to the state.

  • resource_to_mock_fn (Dict[str, Callable]) – A mapping of resource URIs to mock functions to use if the state performs a task.

Returns

The output of the state after applying any output processing.

Return type

Any

class awsstepfuncs.abstract_state.AbstractResultSelectorState(*args, result_selector=None, **kwargs)

Bases: awsstepfuncs.abstract_state.AbstractParametersState

An Amazon States Language state including ResultSelector.

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

simulate(state_input, resource_to_mock_fn)

Simulate the state including input and output processing.

Parameters
  • state_input (Any) – The input to the state.

  • resource_to_mock_fn (Dict[str, Callable]) – A mapping of resource URIs to mock functions to use if the state performs a task.

Returns

The output of the state after applying any output processing.

Return type

Any

class awsstepfuncs.abstract_state.AbstractRetryCatchState(*args, **kwargs)

Bases: awsstepfuncs.abstract_state.AbstractResultSelectorState

An Amazon States Language state including Retry and Catch.

add_catcher(error_equals, *, next_state)

Add a Catcher to the state.

When a state reports an error and either there is no Retrier, or retries have failed to resolve the error, the interpreter will try to find a relevant Catcher which determines which state to transition to.

“States.ALL” is the catch-all error message; that is, any error will be caught with “States.ALL”. Right now it’s the only error code supported when simulating.

If no catcher can be applied, then the state machine will terminate.

Parameters
Returns

Itself to allow for chaining.

Return type

awsstepfuncs.abstract_state.AbstractRetryCatchState

add_retrier(error_equals, *, interval_seconds=None, backoff_rate=None, max_attempts=None)

Add a Retrier to the state.

Retry the state by specifying a list of Retriers that describes the retry policy for different errors.

Parameters
  • error_equals (List[str]) – A list of error names.

  • interval_seconds (Optional[int]) – The number of seconds before the first retry attempt. Defaults to 1 if not specified.

  • backoff_rate (Optional[float]) – A number which is the multiplier that increases the retry interval on each attempt. Defaults to 2.0 if not specified.

  • max_attempts (Optional[int]) – The maximum number of retry to attempt. Defaults to 3 if not specified. A value of zero means that the error should never be retried.

Returns

Itself to allow for chaining.

Return type

awsstepfuncs.abstract_state.AbstractRetryCatchState

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

class awsstepfuncs.abstract_state.AbstractState(name, comment=None)

Bases: abc.ABC

An Amazon States Language state including Name, Comment, and Type.

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

simulate(state_input, *, resource_to_mock_fn)

Simulate the state including input and output processing.

Parameters
  • state_input (Any) – The input to the state.

  • resource_to_mock_fn (Dict[str, Callable]) – A mapping of resource URIs to mock functions to use if the state performs a task.

Returns

The output of the state after applying any output processing.

Return type

Any

awsstepfuncs.choice module

class awsstepfuncs.choice.AbstractChoice(next_state)

Bases: abc.ABC

Choices for Choice State.

evaluate(data)

Evaulate the choice on some given data.

Parameters

data (Any) – Input data to evaluate.

Raises

NotImplementedError – Raised if not implemented in child classes.

Return type

bool

class awsstepfuncs.choice.AndChoice(choice_rules, *, next_state)

Bases: awsstepfuncs.choice.AbstractChoice

And Choice for the Choice State.

The And Choice can be evaluated based on input data to true or false based on whether all Choice Rules are true.

evaluate(data)

Evaulate the And Choice on some given data.

Parameters

data (Any) – Input data to evaluate.

Returns

Whether the choice evaluates to true based on the input data.

Return type

bool

class awsstepfuncs.choice.ChoiceRule(variable, **data_test_expression)

Bases: object

Choice Rules are used in Choices.

When initializing a Choice Rule, a data test expression must be provided. A Choice Rule evalulates to True or False based on the data-test expression on some data.

evaluate(data)

Evaulate the Choice Rule with a data-test expression on some data.

Parameters

data (Any) – Input data to evaluate.

Returns

True or false based on the data and the Choice Rule.

Return type

bool

class awsstepfuncs.choice.DataTestExpression(type, expression)

Bases: object

A data-test expression.

class awsstepfuncs.choice.DataTestExpressionType(value)

Bases: enum.Enum

All the different types of data-test expressions.

Check section “Data-test expression” for a full list: https://states-language.net/#choice-state

BOOLEAN_EQUALS = 'boolean_equals'
BOOLEAN_EQUALS_PATH = 'boolean_equals_path'
IS_BOOLEAN = 'is_boolean'
IS_NULL = 'is_null'
IS_NUMERIC = 'is_numeric'
IS_PRESENT = 'is_present'
IS_STRING = 'is_string'
IS_TIMESTAMP = 'is_timestamp'
NUMERIC_EQUALS = 'numeric_equals'
NUMERIC_EQUALS_PATH = 'numeric_equals_path'
NUMERIC_GREATER_THAN = 'numeric_greater_than'
NUMERIC_GREATER_THAN_EQUALS = 'numeric_greater_than_equals'
NUMERIC_GREATER_THAN_EQUALS_PATH = 'numeric_greater_than_equals_path'
NUMERIC_GREATER_THAN_PATH = 'numeric_greater_than_path'
NUMERIC_LESS_THAN = 'numeric_less_than'
NUMERIC_LESS_THAN_EQUALS = 'numeric_less_than_equals'
NUMERIC_LESS_THAN_EQUALS_PATH = 'numeric_less_than_equals_path'
NUMERIC_LESS_THAN_PATH = 'numeric_less_than_path'
STRING_EQUALS = 'string_equals'
STRING_EQUALS_PATH = 'string_equals_path'
STRING_GREATER_THAN = 'string_greater_than'
STRING_GREATER_THAN_EQUALS = 'string_greater_than_equals'
STRING_GREATER_THAN_EQUALS_PATH = 'string_greater_than_equals_path'
STRING_GREATER_THAN_PATH = 'string_greater_than_path'
STRING_LESS_THAN = 'string_less_than'
STRING_LESS_THAN_EQUALS = 'string_less_than_equals'
STRING_LESS_THAN_EQUALS_PATH = 'string_less_than_equals_path'
STRING_LESS_THAN_PATH = 'string_less_than_path'
STRING_MATCHES = 'string_matches'
TIMESTAMP_EQUALS = 'timestamp_equals'
TIMESTAMP_EQUALS_PATH = 'timestamp_equals_path'
TIMESTAMP_GREATER_THAN = 'timestamp_greater_than'
TIMESTAMP_GREATER_THAN_EQUALS = 'timestamp_greater_than_equals'
TIMESTAMP_GREATER_THAN_EQUALS_PATH = 'timestamp_greater_than_equals_path'
TIMESTAMP_GREATER_THAN_PATH = 'timestamp_greater_than_path'
TIMESTAMP_LESS_THAN = 'timestamp_less_than'
TIMESTAMP_LESS_THAN_EQUALS = 'timestamp_less_than_equals'
TIMESTAMP_LESS_THAN_EQUALS_PATH = 'timestamp_less_than_equals_path'
TIMESTAMP_LESS_THAN_PATH = 'timestamp_less_than_path'
class awsstepfuncs.choice.NotChoice(variable, *, next_state, **data_test_expression)

Bases: awsstepfuncs.choice.AbstractChoice

Not choice for the Choice State.

The Not Choice can be evaluated based on input data to true or false based on whether the Choice Rule is false.

evaluate(data)

Evaulate the Not Choice on some given data.

Parameters

data (Any) – Input data to evaluate.

Returns

Whether the choice evaluates to true based on the input data.

Return type

bool

class awsstepfuncs.choice.VariableChoice(variable, *, next_state, **data_test_expression)

Bases: awsstepfuncs.choice.AbstractChoice

Variable Choice for the Choice State.

The Variable Choice can be evaluated based on input data to true or false based on whether the Choice Rule is true.

Be careful if you use a Reference Path that it evaluates to the correct type.

evaluate(data)

Evaulate the Variable Choice on some given data.

Parameters

data (Any) – Input data to evaluate.

Returns

Whether the choice evaluates to true based on the input data.

Return type

bool

awsstepfuncs.error_handlers module

class awsstepfuncs.error_handlers.AbstractErrorHandler(error_equals)

Bases: abc.ABC

Error handlers compose of Retriers and Catchers.

compile()

Compile the error handler with error_equals handled.

Returns

A compilation with error_equals compiled.

Return type

Dict[str, Any]

class awsstepfuncs.error_handlers.Catcher(error_equals, next_state)

Bases: awsstepfuncs.error_handlers.AbstractErrorHandler

Used to go from an errored state to another state.

compile()

Compile the Catcher to Amazon States Language.

Returns

A Catcher in Amazon States Language.

Return type

Dict[str, Union[List[str], str]]

class awsstepfuncs.error_handlers.Retrier(error_equals, interval_seconds=None, backoff_rate=None, max_attempts=None)

Bases: awsstepfuncs.error_handlers.AbstractErrorHandler

Used to retry a failed state given the error names.

compile()

Compile the Retrier to Amazon States Language.

Returns

A Retrier in Amazon States Language.

Return type

Dict[str, Union[List[str], int, float]]

awsstepfuncs.errors module

exception awsstepfuncs.errors.AWSStepFuncsError

Bases: Exception

Base error for this package.

This is useful for client code to know whether the error is expected or whether there is a bug in the library code.

>>> from awsstepfuncs import *
>>> try:
...     wait_state = WaitState("Wait!", seconds=-1)
... except AWSStepFuncsError:
...     print("Error in the client")
... except Exception:
...     print("Error in awsstepfuncs")
Error in the client
exception awsstepfuncs.errors.AWSStepFuncsValueError

Bases: awsstepfuncs.errors.AWSStepFuncsError

A bad value was specified.

>>> from awsstepfuncs import *
>>> wait_state = WaitState("Wait!", seconds=-1)
Traceback (most recent call last):
    ...
awsstepfuncs.errors.AWSStepFuncsValueError: seconds must be greater than zero
exception awsstepfuncs.errors.FailStateError(*, error, cause)

Bases: awsstepfuncs.errors.StateSimulationError

Raised when running a Fail State.

Parameters
  • error (str) –

  • cause (str) –

exception awsstepfuncs.errors.NoChoiceMatchedError

Bases: awsstepfuncs.errors.StateSimulationError

Raised when a Choice State failed to pick a next state.

error_string = 'States.NoChoiceMatched'
exception awsstepfuncs.errors.StateSimulationError

Bases: awsstepfuncs.errors.AWSStepFuncsError

Raised when there is an error while simulating a state.

Check this table for a list of state simulation errors: https://states-language.net/spec.html#appendix-a

error_string = 'States.ALL'
static from_string(error_string)

Convert a state simulation error string to an error class.

>>> StateSimulationError.from_string("States.Timeout")
<class 'awsstepfuncs.errors.StateTimeoutError'>

If no value is returned, then the error cannot be simulated.

>>> StateSimulationError.from_string("States.Permissions") is None
True
Parameters

error_string (str) – The error string, such as “States.ALL”.

Returns

The error class if the error can be simulated.

Return type

Optional[Type[awsstepfuncs.errors.StateSimulationError]]

exception awsstepfuncs.errors.StateTimeoutError

Bases: awsstepfuncs.errors.StateSimulationError

Raised when the state has timed out while simulating.

A Task State either ran longer than the “TimeoutSeconds” value, or failed to heartbeat for a time longer than the “HeartbeatSeconds” value.

error_string = 'States.Timeout'
exception awsstepfuncs.errors.TaskFailedError

Bases: awsstepfuncs.errors.StateSimulationError

Raised when the task has failed during the execution.

error_string = 'States.TaskFailed'

awsstepfuncs.printer module

class awsstepfuncs.printer.Color(value)

Bases: enum.Enum

Wrap Colorama’s colors to make typing easier for the user.

Refs: https://github.com/tartley/colorama/blob/master/colorama/ansi.py#L49 (AnsiFore class)

BLACK = 1
BLUE = 5
CYAN = 7
GREEN = 3
LIGHTBLACK_EX = 9
LIGHTBLUE_EX = 13
LIGHTCYAN_EX = 15
LIGHTGREEN_EX = 11
LIGHTMAGENTA_EX = 14
LIGHTRED_EX = 10
LIGHTWHITE_EX = 16
LIGHTYELLOW_EX = 12
MAGENTA = 6
RED = 2
WHITE = 8
YELLOW = 4
class awsstepfuncs.printer.Printer(colorful=False)

Bases: object

Print simulation messages to STDOUT.

class awsstepfuncs.printer.Style(value)

Bases: enum.Enum

Wrap Colorama’s styles to make typing easier for the user.

BRIGHT = 2
DIM = 1

awsstepfuncs.reference_path module

class awsstepfuncs.reference_path.ReferencePath(reference_path, /)

Bases: object

Reference Path validation and application.

Reference Path is a specialized JSONPath. Unlike a JSONPath, the Reference Path must be unambiguous and evaluate to only a single node.

More on Reference Paths: https://states-language.net/spec.html#ref-paths More on JSONPath: https://github.com/json-path/JsonPath

apply(data)

Parse then apply a Reference Path on some data.

Parameters

data (dict) – The data to use the Reference Path expression on.

Returns

The queried data.

Return type

Any

awsstepfuncs.state module

State definitions.

Class structure based on this table: https://states-language.net/spec.html#state-type-table

Each row in the table is its own ABC. All latter rows inherit from previous rows. States (the columns) are concrete classes that inherit from the ABC that has the right fields available.

Each concrete class should implement its own _execute() method that will Execute the state according to its business logic when running a simulation. Each concrete class should also define a constant class variable called state_type that corresponds to type in Amazon States Language.

There are two interesting methods common for many classes: - simulate(): Simulate the state including input/output processing - _execute(): Execute the state, eg. for a Wait State, wait the designated time

class awsstepfuncs.state.ChoiceState(*args, choices, default=None, **kwargs)

Bases: awsstepfuncs.state.TerminalStateMixin, awsstepfuncs.abstract_state.AbstractInputPathOutputPathState

A Choice State adds branching logic to a state machine.

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

state_type = 'Choice'
class awsstepfuncs.state.FailState(*args, error, cause, **kwargs)

Bases: awsstepfuncs.state.TerminalStateMixin, awsstepfuncs.abstract_state.AbstractState

The Fail State terminates the machine and marks it as a failure.

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

state_type = 'Fail'
class awsstepfuncs.state.MapState(*args, iterator, items_path='$', max_concurrency, **kwargs)

Bases: awsstepfuncs.abstract_state.AbstractRetryCatchState

The Map State processes all the elements of an array.

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

state_type = 'Map'
class awsstepfuncs.state.ParallelState(*args, **kwargs)

Bases: awsstepfuncs.abstract_state.AbstractRetryCatchState

The Parallel State causes parallel execution of branches.

state_type = 'Parallel'
class awsstepfuncs.state.PassState(*args, result=None, **kwargs)

Bases: awsstepfuncs.abstract_state.AbstractParametersState

The Pass State by default passes its input to its output, performing no work.

If result is passed, its value is treated as the output of a virtual task.

If result_path is specified, the result will be placed on that Reference Path.

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

state_type = 'Pass'
class awsstepfuncs.state.SucceedState(*args, input_path='$', output_path='$', **kwargs)

Bases: awsstepfuncs.state.TerminalStateMixin, awsstepfuncs.abstract_state.AbstractInputPathOutputPathState

The Succeed State terminates with a mark of success.

The Succeed State’s output is the same as its input.

state_type = 'Succeed'
class awsstepfuncs.state.TaskState(*args, resource, timeout_seconds=None, **kwargs)

Bases: awsstepfuncs.abstract_state.AbstractRetryCatchState

The Task State executes the work identified by the Resource field.

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

state_type = 'Task'
class awsstepfuncs.state.TerminalStateMixin

Bases: abc.ABC

A mixin for blocking rshift for terminal states.

class awsstepfuncs.state.WaitState(*args, seconds=None, timestamp=None, seconds_path=None, timestamp_path=None, **kwargs)

Bases: awsstepfuncs.abstract_state.AbstractNextOrEndState

A Wait State causes the interpreter to delay the machine for a specified time.

You can specify a timestamp to wait until. If the time has already past, then there is no wait.

Alternatively, you can use state input to specify the number of seconds wait by specifying a Reference Path seconds_path.

Similarily, you can use state input to specify the timestamp (in ISO 8601 format) to wait until.

Exactly one must be defined: seconds, timestamp, seconds_path, timestamp_path.

Refs: https://states-language.net/#wait-state

compile()

Compile the state to Amazon States Language.

Returns

A dictionary representing the compiled state in Amazon States Language.

Return type

Dict[str, Any]

state_type = 'Wait'

awsstepfuncs.state_machine module

class awsstepfuncs.state_machine.StateMachine(*, start_state, comment=None, version=None)

Bases: object

An AWS Step Functions state machine.

property all_states

Return all states in the state machine.

Returns

A set of all possible states in the state machine.

compile()

Compile a state machine to Amazon States Language.

Returns

A dictionary of the compiled state in Amazon States Language.

Return type

Dict[str, Any]

simulate(state_input=None, /, *, resource_to_mock_fn=None, show_visualization=False, visualization_output_path='state_machine.gif', colorful=False)

Simulate the state machine by executing all of the states.

Parameters
  • state_input (Optional[dict]) – Data to pass to the first state.

  • resource_to_mock_fn (Optional[Dict[str, Callable]]) – A dictionary mapping Resource URI to a mock function to use in the simulation.

  • show_visualization (bool) – Whether or not to create an animated GIF visualization of the state machine when simulating. Outputs to state_machine.gif.

  • visualization_output_path (str) – If show_visualization is set to True, what path to save the visualization GIF to.

  • colorful (bool) – Whether to make the simulation STDOUT messages ✨pop✨.

Returns

The final output state from simulating the state machine.

Return type

Any

to_json(filename)

Compile to Amazon States Language and then output to JSON.

Parameters

filename (Union[str, pathlib.Path]) – The name of the file to write the JSON to.

Return type

None

awsstepfuncs.types module

awsstepfuncs.visualization module

class awsstepfuncs.visualization.Visualization(*, start_state, output_path='state_machine.gif')

Bases: object

Create a visualization of a state machine.

Here’s a few examples of visualizations using show_visualization=True:

resource = "123"
task_state = TaskState("My task", resource=resource)
succeed_state = SucceedState("Success")
pass_state = PassState("Just passing")
fail_state = FailState("Failure", error="IFailed", cause="I failed!")

task_state >> succeed_state
pass_state >> fail_state

task_state.add_catcher(["States.ALL"], next_state=pass_state)

state_machine = StateMachine(start_state=task_state)

def failure_mock_fn(event, context):
    assert False

state_machine.simulate(
    resource_to_mock_fn={resource: failure_mock_fn}, show_visualization=True
)
../_images/state_machine.gif

state_machine.gif

A choice state:

public_state = PassState("Public")
value_in_twenties_state = PassState("ValueInTwenties")
after_value_in_twenties_state = SucceedState("Success!")
start_audit_state = PassState("StartAudit")

value_in_twenties_state >> after_value_in_twenties_state

choice_state = ChoiceState(
    "DispatchEvent",
    choices=[
        NotChoice(
            variable="$.type",
            string_equals="Private",
            next_state=public_state,
        ),
        AndChoice(
            [
                ChoiceRule(variable="$.value", is_present=True),
                ChoiceRule(variable="$.value", numeric_greater_than_equals=20),
                ChoiceRule(variable="$.value", numeric_less_than=30),
            ],
            next_state=value_in_twenties_state,
        ),
        VariableChoice(
            variable="$.rating",
            numeric_greater_than_path="$.auditThreshold",
            next_state=start_audit_state,
        )
    ],
)

state_machine = StateMachine(start_state=choice_state)
state_machine.simulate(
    {"type": "Private", "value": 22}, show_visualization=True
)
../_images/choice_visualization.gif

state_machine.gif

A choice state when the default is chosen:

record_event_state = PassState("RecordEvent")

choice_state = ChoiceState(
    "DispatchEvent",
    choices=[
        NotChoice(
            variable="$.type",
            string_equals="Private",
            next_state=public_state,
        ),
        AndChoice(
            [
                ChoiceRule(variable="$.value", is_present=True),
                ChoiceRule(variable="$.value", numeric_greater_than_equals=20),
                ChoiceRule(variable="$.value", numeric_less_than=30),
            ],
            next_state=value_in_twenties_state,
        ),
        VariableChoice(
            variable="$.rating",
            numeric_greater_than_path="$.auditThreshold",
            next_state=start_audit_state,
        )
    ],
    default=record_event_state,
)
state_machine = StateMachine(start_state=choice_state)
state_machine.simulate(
    {"type": "Private", "value": 102, "auditThreshold": 150},
    show_visualization=True
)
../_images/default_choice_visualization.gif

state_machine.gif

highlight_state(state)

Highlight a state.

Parameters

state (awsstepfuncs.abstract_state.AbstractState) – The state to highlight, unique by name.

Return type

None

highlight_state_transition(previous_state, next_state)

Highlight the transition between two states.

Parameters
Return type

None

render()

Render the state machine visualization to a GIF file.

Return type

None