sfini.state_machine module

State-machine interfacing.

A state-machine defines the logic for a workflow of an application. It is comprised of states (ie stages), and executions of which will run the workflow over some given data.

class sfini.state_machine.StateMachine(name: str, states: Dict[str, sfini.state._base.State], start_state: str, comment: str = DefaultParameter(), timeout: int = DefaultParameter(), *, session: sfini._util.AWSSession = None)[source]

Bases: object

State machine structure for AWS Step Functions.

Parameters:
  • name – name of state-machine
  • states – state-machine states
  • start_state – name of start state
  • comment – description of state-maching
  • timeout – execution time-out (seconds)
  • session – session to use for AWS communication
arn

State-machine generated ARN.

default_role_arn

sfini-generated state-machine IAM role ARN.

deregister()[source]

Remove state-machine from AWS SFN.

is_registered() → bool[source]

See if this state-machine is registered with AWS SFN.

Returns:if this state-machine is registered
list_executions(status: str = None) → List[sfini.execution._execution.Execution][source]

List all executions of this state-machine.

This state-machine is manually attached to the state_machine attribute of the resultant executions here.

Parameters:status – only list executions with this status. Choose from ‘RUNNING’, ‘SUCCEEDED’, ‘FAILED’, ‘TIMED_OUT’ or ‘ABORTED’
Returns:executions of this state-machine
register(role_arn: str = None, allow_update: bool = False)[source]

Register state-machine with AWS SFN.

Parameters:
  • role_arn – state-machine IAM role ARN
  • allow_update – allow overwriting of an existing state-machine with the same name
start_execution(execution_input: Union[None, bool, str, int, float, List[JSONable], Dict[str, JSONable]]) → sfini.execution._execution.Execution[source]

Start an execution.

Parameters:execution_input – input to first state in state-machine
Returns:started execution
to_dict() → Dict[str, Union[None, bool, str, int, float, List[JSONable], Dict[str, JSONable]]][source]

Convert this state-machine to a definition dictionary.

Returns:definition
sfini.state_machine.construct_state_machine(name: str, start_state: sfini.state._base.State, comment: str = DefaultParameter(), timeout: int = DefaultParameter(), *, session: sfini._util.AWSSession = None) → sfini.state_machine.StateMachine[source]

Construct a state-machine from the starting state.

Make sure to construct the state-machine after all states have been defined: subsequent states will need to be added to the state-machine manually.

Only states referenced by the provided first state (and their children) will be in the state-machine definition. Add states via an impossible choice rule to include them in the definition.

Parameters:
  • name – name of state-machine
  • start_state – starting state of state-machine
  • comment – description of state-maching
  • timeout – execution time-out (seconds)
  • session – session to use for AWS communication
Returns:

constructed state-machine