Step Functions States¶

Standard Step Functions and state machine states.

See Step Functions docs for more details.

class rhodes.states.State(title, *, Comment=None)[source]¶

Bases: object

Base class for states.

member_of = None¶

to_dict()[source]¶

Serialize state as a dictionary.

Return type:	`Dict`

promote(path)[source]¶

Add a Pass state after this state that promotes a path in the input to this state’s ResultPath.

Path must start with a path relative this state’s ResultPath as indicated by a @. prefix.

Parameters:	path (`Union`[`str`, `Enum`, `JSONPath`, `JsonPath`]) – Path to promote
Return type:	`Pass`

class rhodes.states.StateMachine(*, States=NOTHING, StartAt=None, Comment=None, Version=None, TimeoutSeconds=None)[source]¶

Bases: object

Step Functions State Machine.

See Step Functions docs for more details.

Parameters:

States (dict(str, State)) – Map of states that make up this state machine
StartAt (str) – The state where this state machine starts
Comment (str) – Human-readable description of the state
Version (str) – The version of the Amazon States Language used in this state machine (must be 1.0 if provided)
TimeoutSeconds (int) – Maximum time that this state machine is allowed to run

to_dict()[source]¶

Serialize this state machine as a dictionary.

Return type:	`Dict`

definition_string()[source]¶

Serialize this state machine for use in a troposphere state machine definition.

Return type:	`Sub`

add_state(new_state)[source]¶

Add a state to this state machine.

Parameters:	new_state (State) – State to add
Return type:	`State`

start_with(first_state)[source]¶

Add a state to this state machine and mark it as the starting state.

Parameters:	first_state (State) – State to start with
Return type:	`State`

class rhodes.states.Pass(title, *, Comment=None, Result=None, Next=None, End=None, InputPath=JsonPath(path=Root()), OutputPath=JsonPath(path=Root()), ResultPath=JsonPath(path=Root()), Parameters=None)[source]¶

Bases: rhodes.states.State

A Pass state passes its input to its output without performing work. Pass states are useful when constructing and debugging state machines.

See Step Functions docs for more details.

Parameters:

title (str) – Name of state in state machine
Comment (str) – Human-readable description of the state (default: '')
Next – The state that will follow this state
End (bool) – This state is a terminal state
InputPath (JsonPath) – The portion of the state input data to be used as input for the state (default: JsonPath(path=Root()))
OutputPath (JsonPath) – The portion of the state input data to be passed to the next state (default: JsonPath(path=Root()))
ResultPath (JsonPath) – Where in the state input data to place the results of this state (default: JsonPath(path=Root()))
Parameters (Parameters) – Additional parameters for Step Functions to provide to connected resource

end()¶: Make this state a terminal state.

member_of = None¶

promote(path)¶

Add a Pass state after this state that promotes a path in the input to this state’s ResultPath.

Path must start with a path relative this state’s ResultPath as indicated by a @. prefix.

Parameters:	path (`Union`[`str`, `Enum`, `JSONPath`, `JsonPath`]) – Path to promote
Return type:	`Pass`

then(next_state)¶: Set the next state in this state machine.

to_dict()¶

Serialize state as a dictionary.

Return type:	`Dict`

class rhodes.states.Task(title, *, Comment=None, Resource=None, Next=None, End=None, InputPath=JsonPath(path=Root()), OutputPath=JsonPath(path=Root()), ResultPath=JsonPath(path=Root()), Retry=None, Catch=None, TimeoutSeconds=None, HeartbeatSeconds=None, Parameters=None)[source]¶

Bases: rhodes.states.State

A Task state represents a single unit of work performed by a state machine.

See Step Functions docs for more details.

Parameters:

title (str) – Name of state in state machine
Comment (str) – Human-readable description of the state (default: '')
Next – The state that will follow this state
End (bool) – This state is a terminal state
InputPath (JsonPath) – The portion of the state input data to be used as input for the state (default: JsonPath(path=Root()))
OutputPath (JsonPath) – The portion of the state input data to be passed to the next state (default: JsonPath(path=Root()))
ResultPath (JsonPath) – Where in the state input data to place the results of this state (default: JsonPath(path=Root()))
Retry –
Catch –
TimeoutSeconds (int) – Maximum time that this state is allowed to run
HeartbeatSeconds (int) – Maximum time allowed between heartbeat responses from state
Parameters (Parameters) – Additional parameters for Step Functions to provide to connected resource

end()¶: Make this state a terminal state.

member_of = None¶

promote(path)¶

Add a Pass state after this state that promotes a path in the input to this state’s ResultPath.

Path must start with a path relative this state’s ResultPath as indicated by a @. prefix.

Parameters:	path (`Union`[`str`, `Enum`, `JSONPath`, `JsonPath`]) – Path to promote
Return type:	`Pass`

then(next_state)¶: Set the next state in this state machine.

to_dict()¶

Serialize state as a dictionary.

Return type:	`Dict`

class rhodes.states.Choice(title, *, Comment=None, Choices=NOTHING, Default=None, InputPath=JsonPath(path=Root()), OutputPath=JsonPath(path=Root()))[source]¶

Bases: rhodes.states.State

A Choice state adds branching logic to a state machine.

See Step Functions docs for more details.

workflow = StateMachine()

next_state = Pass("Ok number")

decision = workflow.start_with(Choice("Make a decision"))
decision.if_(VariablePath("$.foo.bar") == 12).then(next_state)
decision.if_(VariablePath("$.foo.bar") < 0).then(Fail("Negative value!"))
decision.if_(all_(
    VariablePath("$.foo.bar") != 12,
    VariablePath("$.baz.wow") == "not 12!",
)).then(next_state)
decision.else_(Succeed("Something else!"))

next_state.end()

{
    "States": {
        "Make a decision": {
            "Type": "Choice",
            "Choices": [
                {
                    "Variable": "$.foo.bar",
                    "NumericEquals": 12,
                    "Next": "Ok number"
                },
                {
                    "Variable": "$.foo.bar",
                    "NumericLessThan": 0,
                    "Next": "Negative value!"
                },
                {
                    "And": [
                        {
                            "Not": {
                                "Variable": "$.foo.bar",
                                "NumericEquals": 12
                            }
                        },
                        {
                            "Variable": "$.baz.wow",
                            "StringEquals": "not 12!"
                        }
                    ],
                    "Next": "Ok number"
                }
            ],
            "Default": "Something else!"
        },
        "Ok number": {
            "Type": "Pass",
            "End": true
        },
        "Negative value!": {
            "Type": "Fail"
        },
        "Something else!": {
            "Type": "Succeed"
        }
    }
}

Parameters:

title (str) – Name of state in state machine
Comment (str) – Human-readable description of the state (default: '')
InputPath (JsonPath) – The portion of the state input data to be used as input for the state (default: JsonPath(path=Root()))
OutputPath (JsonPath) – The portion of the state input data to be passed to the next state (default: JsonPath(path=Root()))

add_choice(rule)[source]¶

Add a choice rule to this state. This is the lower-level interface that :method:`if_` uses.

Parameters:	rule (`ChoiceRule`) – Rule to add
Return type:	`ChoiceRule`
Returns:	`rule`

if_(rule)[source]¶

Add a choice rule to this state as one possible logic branch. This should be followed up with a .then(STATE) call.

decision.if_(VariablePath("$.foo.bar") == 12).then(next_state)

This results in the rule definition:

{
    "Variable": "$.foo.bar",
    "NumericEquals": 12,
    "Next": "NEXT_STATE_NAME"
}

Parameters:	rule (`ChoiceRule`) – The rule to add
Return type:	`ChoiceRule`
Returns:	`rule`

else_(state)[source]¶

Add a default state. This is the state to transition to if none of the choice rules are satisfied.

Parameters:	state (`State`) – The default state to add
Return type:	`State`
Returns:	`state`

to_dict()[source]¶

Serialize state as a dictionary.

Return type:	`Dict`

member_of = None¶

promote(path)¶

Add a Pass state after this state that promotes a path in the input to this state’s ResultPath.

Path must start with a path relative this state’s ResultPath as indicated by a @. prefix.

Parameters:	path (`Union`[`str`, `Enum`, `JSONPath`, `JsonPath`]) – Path to promote
Return type:	`Pass`

class rhodes.states.Wait(title, *, Comment=None, Seconds=None, Timestamp=None, SecondsPath=None, TimestampPath=None, Next=None, End=None, InputPath=JsonPath(path=Root()), OutputPath=JsonPath(path=Root()))[source]¶

Bases: rhodes.states.State

A Wait state delays the state machine from continuing for a specified time. You can choose either a relative time, specified in seconds from when the state begins, or an absolute end time, specified as a timestamp.

See Step Functions docs for more details.

Parameters:

title (str) – Name of state in state machine
Comment (str) – Human-readable description of the state (default: '')
Next – The state that will follow this state
End (bool) – This state is a terminal state
InputPath (JsonPath) – The portion of the state input data to be used as input for the state (default: JsonPath(path=Root()))
OutputPath (JsonPath) – The portion of the state input data to be passed to the next state (default: JsonPath(path=Root()))

end()¶: Make this state a terminal state.

member_of = None¶

promote(path)¶

Add a Pass state after this state that promotes a path in the input to this state’s ResultPath.

Path must start with a path relative this state’s ResultPath as indicated by a @. prefix.

Parameters:	path (`Union`[`str`, `Enum`, `JSONPath`, `JsonPath`]) – Path to promote
Return type:	`Pass`

then(next_state)¶: Set the next state in this state machine.

to_dict()¶

Serialize state as a dictionary.

Return type:	`Dict`

class rhodes.states.Succeed(title, *, Comment=None)[source]¶

Bases: rhodes.states.State

A Succeed state stops an execution successfully. The Succeed state is a useful target for Choice state branches that don’t do anything but stop the execution.

See Step Functions docs for more details.

Parameters:	title (str) – Name of state in state machine Comment (str) – Human-readable description of the state (default: `''`)

member_of = None¶

promote(path)¶

Add a Pass state after this state that promotes a path in the input to this state’s ResultPath.

Path must start with a path relative this state’s ResultPath as indicated by a @. prefix.

Parameters:	path (`Union`[`str`, `Enum`, `JSONPath`, `JsonPath`]) – Path to promote
Return type:	`Pass`

to_dict()¶

Serialize state as a dictionary.

Return type:	`Dict`

class rhodes.states.Fail(title, *, Comment=None, Error=None, Cause=None)[source]¶

Bases: rhodes.states.State

A Fail state stops the execution of the state machine and marks it as a failure.

See Step Functions docs for more details.

Parameters:	title (str) – Name of state in state machine Comment (str) – Human-readable description of the state (default: `''`)

member_of = None¶

promote(path)¶

Add a Pass state after this state that promotes a path in the input to this state’s ResultPath.

Path must start with a path relative this state’s ResultPath as indicated by a @. prefix.

Parameters:	path (`Union`[`str`, `Enum`, `JSONPath`, `JsonPath`]) – Path to promote
Return type:	`Pass`

to_dict()¶

Serialize state as a dictionary.

Return type:	`Dict`

class rhodes.states.Parallel(title, *, Comment=None, Branches=NOTHING, Next=None, End=None, InputPath=JsonPath(path=Root()), OutputPath=JsonPath(path=Root()), ResultPath=JsonPath(path=Root()), Retry=None, Catch=None, Parameters=None)[source]¶

Bases: rhodes.states.State

The Parallel state can be used to create parallel branches of execution in your state machine.

See Step Functions docs for more details.

Parameters:

title (str) – Name of state in state machine
Comment (str) – Human-readable description of the state (default: '')
Next – The state that will follow this state
End (bool) – This state is a terminal state
InputPath (JsonPath) – The portion of the state input data to be used as input for the state (default: JsonPath(path=Root()))
OutputPath (JsonPath) – The portion of the state input data to be passed to the next state (default: JsonPath(path=Root()))
ResultPath (JsonPath) – Where in the state input data to place the results of this state (default: JsonPath(path=Root()))
Retry –
Catch –
Parameters (Parameters) – Additional parameters for Step Functions to provide to connected resource

to_dict()[source]¶

Serialize state as a dictionary.

Return type:	`Dict`

add_branch(state_machine=None)[source]¶

Add a parallel branch to this state. If state_machine is not provided, we generate an empty state machine and add that.

Parameters:	state_machine (`Optional`[`StateMachine`]) – State machine to add (optional)
Return type:	`StateMachine`
Returns:	`state_machine` if provided or a new empty state machine if not

end()¶: Make this state a terminal state.

member_of = None¶

promote(path)¶

Add a Pass state after this state that promotes a path in the input to this state’s ResultPath.

Path must start with a path relative this state’s ResultPath as indicated by a @. prefix.

Parameters:	path (`Union`[`str`, `Enum`, `JSONPath`, `JsonPath`]) – Path to promote
Return type:	`Pass`

then(next_state)¶: Set the next state in this state machine.

class rhodes.states.Map(title, *, Comment=None, Iterator=None, ItemsPath=None, MaxConcurrency=None, Next=None, End=None, InputPath=JsonPath(path=Root()), OutputPath=JsonPath(path=Root()), ResultPath=JsonPath(path=Root()), Retry=None, Catch=None, Parameters=None)[source]¶

Bases: rhodes.states.State

The Map state can be used to run a set of steps for each element of an input array. While the Parallel state executes multiple branches of steps using the same input, a Map state will execute the same steps for multiple entries of an array in the state input.

See Step Functions docs for more details.

Parameters:

title (str) – Name of state in state machine
Comment (str) – Human-readable description of the state (default: '')
Next – The state that will follow this state
End (bool) – This state is a terminal state
InputPath (JsonPath) – The portion of the state input data to be used as input for the state (default: JsonPath(path=Root()))
OutputPath (JsonPath) – The portion of the state input data to be passed to the next state (default: JsonPath(path=Root()))
ResultPath (JsonPath) – Where in the state input data to place the results of this state (default: JsonPath(path=Root()))
Retry –
Catch –
Parameters (Parameters) – Additional parameters for Step Functions to provide to connected resource

end()¶: Make this state a terminal state.

member_of = None¶

promote(path)¶

Add a Pass state after this state that promotes a path in the input to this state’s ResultPath.

Path must start with a path relative this state’s ResultPath as indicated by a @. prefix.

Parameters:	path (`Union`[`str`, `Enum`, `JSONPath`, `JsonPath`]) – Path to promote
Return type:	`Pass`

then(next_state)¶: Set the next state in this state machine.

to_dict()¶

Serialize state as a dictionary.

Return type:	`Dict`