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.

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[[Any], Any]]) – 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 includin 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[[Any], Any]]) – 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[[Any], Any]]) – 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.

Parameters

error_equals (List[str]) – A list of error names.
next_state (awsstepfuncs.abstract_state.AbstractState) – The name of the next state.

Returns

Itself.

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.

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[[Any], Any]]) – 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.Catcher(error_equals, next_state)¶

Bases: object

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]]

error_equals: List[str]¶

next_state: awsstepfuncs.abstract_state.AbstractState¶

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

Bases: object

Used to retry a failed state given the error names.

backoff_rate: Optional[float] = None¶

compile()¶

Compile the Retrier to Amazon States Language.

Returns: A Retrier in Amazon States Language.
Return type: Dict[str, Union[List[str], int, float]]

error_equals: List[str]¶

interval_seconds: Optional[int] = None¶

max_attempts: Optional[int] = None¶

awsstepfuncs.abstract_state.apply_input_path(input_path, state_input)¶

Apply input path to some state input.

Parameters

input_path (awsstepfuncs.json_path.JSONPath) –
state_input (Any) –

Return type

Any

awsstepfuncs.abstract_state.apply_output_path(output_path, state_output)¶

Apply output path to some state output.

Parameters

output_path (awsstepfuncs.json_path.JSONPath) –
state_output (Any) –

Return type

Any

awsstepfuncs.json_path module¶

class awsstepfuncs.json_path.JSONPath(json_path, /)¶

Bases: object

JSONPath validation and application.

See: https://github.com/json-path/JsonPath

apply(data)¶

Parse then apply a JSONPath on some data.

Parameters: data (dict) – The data to use the JSONPath 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 _run() method that will run 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 - _run() – Run the state, eg. for a WaitState wait the designated time

class awsstepfuncs.state.AbstractChoice(next_state)¶

Bases: abc.ABC

Choices for Choice State.

class awsstepfuncs.state.AndChoice(conditions, *, next_state)¶

Bases: awsstepfuncs.state.AbstractChoice

And Choice for the Choice State.

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.

Define some states that can be conditionally transitioned to by the Choice State.

>>> public_state = PassState("Public")
>>> value_in_twenties_state = PassState("ValueInTwenties")
>>> start_audit_state = PassState("StartAudit")
>>> record_event_state = PassState("RecordEvent")

Now we can define a Choice State with branching logic based on conditions.

>>> choice_state = ChoiceState(
...     "DispatchEvent",
...     choices=[
...         NotChoice(
...             variable="$.type",
...             string_equals="Private",
...             next_state=public_state,
...         ),
...         AndChoice(
...             [
...                 Condition(variable="$.value", is_present=True),
...                 Condition(variable="$.value", numeric_greater_than_equals=20),
...                 Condition(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,
... )

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.Condition(variable, *, string_equals=None, is_present=None, numeric_greater_than_equals=None, numeric_greater_than_path=None, numeric_less_than=None)¶

Bases: object

Conditions are used in Choices.

A Condition evalulates to True or False.

>>> career_condition = Condition("$.career", string_equals="Pirate")
>>> career_condition.evaluate({"career": "Pirate", "salary": "10 guineas"})
True
>>> career_condition.evaluate({"career": "Sailor", "salary": "5 guineas"})
False

There can only be one “clause” per condition.

>>> Condition("$.career", string_equals="Pirate", is_present=True)
Traceback (most recent call last):
    ...
ValueError: Exactly one must be defined: string_equals, is_present, numeric_greater_than_equals, numeric_greater_than_path, numeric_less_than

evaluate(data)¶

Evaulate the condition on some given data.

Parameters: data (Any) – Input data to evaluate.
Returns: True or false based on the data and the Condition.
Return type: bool

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.

>>> fail_state = FailState("Failure", error="IFailed", cause="I failed!")
>>> state_machine = StateMachine(start_state=fail_state)
>>> state_output = state_machine.simulate()
Starting simulation of state machine
Running FailState('Failure', error='IFailed', cause='I failed!')
State input: {}
State output: {}
Terminating simulation of state machine

compile()¶

Compile the state to Amazon States Language.

>>> fail_state = FailState("FailState", error="ErrorA", cause="Kaiju attack")
>>> fail_state.compile()
{'Type': 'Fail', 'Error': 'ErrorA', 'Cause': 'Kaiju attack'}

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.

>>> resource = "<arn>"
>>> task_state = TaskState("Validate", resource=resource)
>>> iterator = StateMachine(start_state=task_state)
>>> map_state = MapState(
...     "Validate-All",
...     input_path="$.detail",
...     items_path="$.shipped",
...     max_concurrency=0,
...     iterator=iterator,
... )
>>> state_machine = StateMachine(start_state=map_state)

You can simulate a state machine with a Map State.

>>> state_input = {
...    "ship-date": "2016-03-14T01:59:00Z",
...    "detail": {
...        "delivery-partner": "UQS",
...        "shipped": [
...            {"prod": "R31", "dest-code": 9511, "quantity": 1344},
...            {"prod": "S39", "dest-code": 9511, "quantity": 40},
...        ],
...    },
... }
>>> def mock_fn(state_input):
...     state_input["quantity"] *= 2
...     return state_input
>>> _ = state_machine.simulate(
...     state_input=state_input,
...     resource_to_mock_fn={resource: mock_fn},
... )
Starting simulation of state machine
Running MapState('Validate-All')
State input: {'ship-date': '2016-03-14T01:59:00Z', 'detail': {'delivery-partner': 'UQS', 'shipped': [{'prod': 'R31', 'dest-code': 9511, 'quantity': 1344}, {'prod': 'S39', 'dest-code': 9511, 'quantity': 40}]}}
State input after applying input path of "$.detail": {'delivery-partner': 'UQS', 'shipped': [{'prod': 'R31', 'dest-code': 9511, 'quantity': 1344}, {'prod': 'S39', 'dest-code': 9511, 'quantity': 40}]}
Items after applying items_path of $.shipped: [{'prod': 'R31', 'dest-code': 9511, 'quantity': 1344}, {'prod': 'S39', 'dest-code': 9511, 'quantity': 40}]
Starting simulation of state machine
Running TaskState('Validate')
State input: {'prod': 'R31', 'dest-code': 9511, 'quantity': 1344}
State input after applying input path of "$": {'prod': 'R31', 'dest-code': 9511, 'quantity': 1344}
Output from applying result path of "$": {'prod': 'R31', 'dest-code': 9511, 'quantity': 2688}
State output after applying output path of "$": {'prod': 'R31', 'dest-code': 9511, 'quantity': 2688}
State output: {'prod': 'R31', 'dest-code': 9511, 'quantity': 2688}
Terminating simulation of state machine
Starting simulation of state machine
Running TaskState('Validate')
State input: {'prod': 'S39', 'dest-code': 9511, 'quantity': 40}
State input after applying input path of "$": {'prod': 'S39', 'dest-code': 9511, 'quantity': 40}
Output from applying result path of "$": {'prod': 'S39', 'dest-code': 9511, 'quantity': 80}
State output after applying output path of "$": {'prod': 'S39', 'dest-code': 9511, 'quantity': 80}
State output: {'prod': 'S39', 'dest-code': 9511, 'quantity': 80}
Terminating simulation of state machine
Output from applying result path of "$": [{'prod': 'R31', 'dest-code': 9511, 'quantity': 2688}, {'prod': 'S39', 'dest-code': 9511, 'quantity': 80}]
State output after applying output path of "$": [{'prod': 'R31', 'dest-code': 9511, 'quantity': 2688}, {'prod': 'S39', 'dest-code': 9511, 'quantity': 80}]
State output: [{'prod': 'R31', 'dest-code': 9511, 'quantity': 2688}, {'prod': 'S39', 'dest-code': 9511, 'quantity': 80}]
Terminating simulation of state machine

You can also compile a state machine with a Map State.

>>> output = state_machine.compile()
>>> expected_output = {
...     "StartAt": "Validate-All",
...     "States": {
...         "Validate-All": {
...             "Type": "Map",
...             "InputPath": "$.detail",
...             "End": True,
...             "ItemsPath": "$.shipped",
...             "MaxConcurrency": 0,
...             "Iterator": {
...                 "StartAt": "Validate",
...                 "States": {
...                     "Validate": {"Type": "Task", "End": True, "Resource": "<arn>"}
...                 },
...             },
...         }
...     },
... }
>>> assert output == expected_output

Be careful that items_path Reference Path actually yields a list.

>>> map_state = MapState(
...     "Validate-All",
...     input_path="$.detail",
...     items_path="$.delivery-partner",
...     max_concurrency=0,
...     iterator=iterator,
... )
>>> state_machine = StateMachine(start_state=map_state)
>>> _ = state_machine.simulate(
...     state_input=state_input,
...     resource_to_mock_fn={resource: mock_fn},
... )
Starting simulation of state machine
Running MapState('Validate-All')
State input: {'ship-date': '2016-03-14T01:59:00Z', 'detail': {'delivery-partner': 'UQS', 'shipped': [{'prod': 'R31', 'dest-code': 9511, 'quantity': 2688}, {'prod': 'S39', 'dest-code': 9511, 'quantity': 80}]}}
State input after applying input path of "$.detail": {'delivery-partner': 'UQS', 'shipped': [{'prod': 'R31', 'dest-code': 9511, 'quantity': 2688}, {'prod': 'S39', 'dest-code': 9511, 'quantity': 80}]}
Items after applying items_path of $.delivery-partner: UQS
Error encountered in state, checking for catchers
State output: {}
Terminating simulation of 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 = 'Map'¶

class awsstepfuncs.state.NotChoice(variable, *, next_state, string_equals=None, is_present=None, numeric_greater_than_equals=None, numeric_greater_than_path=None, numeric_less_than=None)¶

Bases: awsstepfuncs.state.AbstractChoice

Not choice for the Choice State.

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.

>>> result = {"Hello": "world!"}
>>> pass_state = PassState("Passing", result=result)
>>> state_machine = StateMachine(start_state=pass_state)
>>> state_output = state_machine.simulate()
Starting simulation of state machine
Running PassState('Passing')
State input: {}
State input after applying input path of "$": {}
Output from applying result path of "$": {'Hello': 'world!'}
State output after applying output path of "$": {'Hello': 'world!'}
State output: {'Hello': 'world!'}
Terminating simulation of state machine
>>> assert state_output == result

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

>>> result = {"Hello": "world!"}
>>> pass_state = PassState("Passing", result=result, result_path="$.result")
>>> state_machine = StateMachine(start_state=pass_state)
>>> _ = state_machine.simulate(state_input={"sum": 42})
Starting simulation of state machine
Running PassState('Passing')
State input: {'sum': 42}
State input after applying input path of "$": {'sum': 42}
Output from applying result path of "$.result": {'sum': 42, 'result': {'Hello': 'world!'}}
State output after applying output path of "$": {'sum': 42, 'result': {'Hello': 'world!'}}
State output: {'sum': 42, 'result': {'Hello': 'world!'}}
Terminating simulation of state machine

compile()¶

Compile the state to Amazon States Language.

>>> result = {"Hello": "world!"}
>>> pass_state = PassState("Passing", result=result)
>>> pass_state.compile()
{'Type': 'Pass', 'End': True, 'Result': {'Hello': 'world!'}}

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 branch can either be:

The entire state machine
A branch of a Parallel State
An iteration of a Map State

state_type = 'Succeed'¶

class awsstepfuncs.state.TaskState(*args, resource, **kwargs)¶

Bases: awsstepfuncs.abstract_state.AbstractRetryCatchState

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

>>> task_state = TaskState("Task", resource="123").add_retrier(["SomeError"], max_attempts=0)
>>> task_state.compile()
{'Type': 'Task', 'End': True, 'Retry': [{'ErrorEquals': ['SomeError'], 'MaxAttempts': 0}], 'Resource': '123'}

>>> fail_state = FailState("Fail", error="SomeError", cause="I did it!")
>>> _ = task_state >> fail_state

When the state machine simulates the previous example, task_state should not get retried as even though a retrier was set for the thrown error, max attempts set to zero means it will not be retried.

>>> transition_state = TaskState("Cleanup", resource="456")
>>> _ = task_state.add_catcher(["States.ALL"], next_state=transition_state)
>>> task_state.compile()
{'Type': 'Task', 'Next': 'Fail', 'Retry': [{'ErrorEquals': ['SomeError'], 'MaxAttempts': 0}], 'Catch': [{'ErrorEquals': ['States.ALL'], 'Next': 'Cleanup'}], 'Resource': '123'}

>>> another_fail_state = FailState("AnotherFail", error="AnotherError", cause="I did it again!")
>>> _ = task_state >> another_fail_state

When the state machine simulates the previous example, in this case, we should end up at transition_state because “States.ALL” catches all errors and transitions to transition_state.

compile()¶

Compile the state to Amazon States Language.

>>> task_state = TaskState("Task", resource="arn:aws:lambda:ap-southeast-2:710187714096:function:DummyResource")
>>> task_state.compile()
{'Type': 'Task', 'End': True, 'Resource': 'arn:aws:lambda:ap-southeast-2:710187714096:function:DummyResource'}

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.VariableChoice(variable, *, next_state, string_equals=None, is_present=None, numeric_greater_than_equals=None, numeric_greater_than_path=None, numeric_less_than=None)¶

Bases: awsstepfuncs.state.AbstractChoice

Variable Choice for the Choice State.

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 the number of seconds to wait.

>>> wait_state = WaitState("Wait!", seconds=1)
>>> state_machine = StateMachine(start_state=wait_state)
>>> state_output = state_machine.simulate()
Starting simulation of state machine
Running WaitState('Wait!', seconds=1)
State input: {}
State input after applying input path of "$": {}
Waiting 1 seconds
State output after applying output path of "$": {}
State output: {}
Terminating simulation of state machine

Seconds must be an integer greater than zero.

>>> WaitState("Wait!", seconds=-1)
Traceback (most recent call last):
    ...
ValueError: seconds must be greater than zero

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

>>> from datetime import datetime, timedelta
>>> wait_state = WaitState("Wait!", timestamp=datetime(2020, 1, 1))
>>> state_machine = StateMachine(start_state=wait_state)
>>> state_output = state_machine.simulate()
Starting simulation of state machine
Running WaitState('Wait!', timestamp='2020-01-01T00:00:00')
State input: {}
State input after applying input path of "$": {}
State output after applying output path of "$": {}
State output: {}
Terminating simulation of state machine

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

>>> wait_state = WaitState("Wait!", seconds_path="$.numSeconds")
>>> state_machine = StateMachine(start_state=wait_state)
>>> state_output = state_machine.simulate(state_input={"numSeconds": 1})
Starting simulation of state machine
Running WaitState('Wait!', seconds_path='$.numSeconds')
State input: {'numSeconds': 1}
State input after applying input path of "$": {'numSeconds': 1}
Waiting 1 seconds
State output after applying output path of "$": {'numSeconds': 1}
State output: {'numSeconds': 1}
Terminating simulation of state machine

A ValueError will be thrown if seconds_path isn’t a reference path to an integer. This is considered a runtime exception and will be treated as an error during the simulation.

>>> wait_state = WaitState("Wait!", seconds_path="$.numSeconds")
>>> state_machine = StateMachine(start_state=wait_state)
>>> state_output = state_machine.simulate(state_input={"numSeconds": "hello"})
Starting simulation of state machine
Running WaitState('Wait!', seconds_path='$.numSeconds')
State input: {'numSeconds': 'hello'}
State input after applying input path of "$": {'numSeconds': 'hello'}
Error encountered in state, checking for catchers
State output: {}
Terminating simulation of state machine

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

>>> wait_state = WaitState("Wait!", timestamp_path="$.meta.timeToWait")
>>> state_machine = StateMachine(start_state=wait_state)
>>> state_output = state_machine.simulate(state_input={"meta": {"timeToWait": "2020-01-01T00:00:00"}})
Starting simulation of state machine
Running WaitState('Wait!', timestamp_path='$.meta.timeToWait')
State input: {'meta': {'timeToWait': '2020-01-01T00:00:00'}}
State input after applying input path of "$": {'meta': {'timeToWait': '2020-01-01T00:00:00'}}
Waiting until 2020-01-01T00:00:00
State output after applying output path of "$": {'meta': {'timeToWait': '2020-01-01T00:00:00'}}
State output: {'meta': {'timeToWait': '2020-01-01T00:00:00'}}
Terminating simulation of state machine

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

Multiple parameters set:

>>> WaitState("Wait", seconds=5, timestamp=datetime.now())
Traceback (most recent call last):
    ...
ValueError: Exactly one must be defined: seconds, timestamp, seconds_path, timestamp_path

No parameters set:

>>> WaitState("Wait")
Traceback (most recent call last):
    ...
ValueError: 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.

>>> WaitState("Wait!", seconds=5).compile()
{'Type': 'Wait', 'End': True, 'Seconds': 5}

>>> WaitState("Wait!", timestamp=datetime(2020,1,1)).compile()
{'Type': 'Wait', 'End': True, 'Timestamp': '2020-01-01T00:00:00'}

>>> WaitState("Wait!", seconds_path="$.numSeconds").compile()
{'Type': 'Wait', 'End': True, 'SecondsPath': '$.numSeconds'}

>>> WaitState("Wait!", timestamp_path="$.meta.timeToWait").compile()
{'Type': 'Wait', 'End': True, 'TimestampPath': '$.meta.timeToWait'}

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)¶

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[[Any], Any]]]) – 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.

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)¶

Bases: object

Create a visualization of a state machine.

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

previous_state (awsstepfuncs.abstract_state.AbstractState) – The previous state.
next_state (awsstepfuncs.abstract_state.AbstractState) – The next state.

Return type

None

render()¶

Render the state machine visualization to a file (state_machine.gif).

Return type: None