sfini.state package

Submodules

• sfini.state.choice module

Module contents

State definitions.

States comprise a state-machine, defining its logic and which activities to run, and direct data.

class sfini.state.State(name: str, comment: str = DefaultParameter(), input_path: Optional[str] = DefaultParameter(), output_path: Optional[str] = DefaultParameter())

Bases: object

Abstract state.

Parameters:

• name – name of state
• comment – state description
• input_path – state input filter JSONPath, None for empty input
• output_path – state output filter JSONPath, None for discarded output

add_to(states)

Add this state to a state-machine definition.

Any child states will also be added to the definition.

Parameters:states (dict[str, State]) – state-machine states

to_dict() → Dict[str, Union[None, bool, str, int, float, List[JSONable], Dict[str, JSONable]]]

Convert this state to a definition dictionary.

Returns:definition

class sfini.state.HasNext(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter())

Bases: sfini.state._base.State

State able to advance mix-in.

Parameters:

• name – name of state
• comment – state description
• input_path – state input filter JSONPath, None for empty input
• output_path – state output filter JSONPath, None for discarded output

next

next state to execute, or None if state is terminal

add_to(states)

Add this state to a state-machine definition.

Any child states will also be added to the definition.

Parameters:states (dict[str, State]) – state-machine states

goes_to(state: sfini.state._base.State)

Set next state after this state finishes.

Parameters:state – state to execute next

remove_next()

Remove next state, making this state terminal.

to_dict()

Convert this state to a definition dictionary.

Returns:definition

class sfini.state.HasResultPath(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter(), result_path: Optional[str] = DefaultParameter())

Bases: sfini.state._base.State

State with result mix-in.

Parameters:

• name – name of state
• comment – state description
• input_path – state input filter JSONPath, None for empty input
• output_path – state output filter JSONPath, None for discarded output
• result_path – task output location JSONPath, None for discarded output

to_dict()

Convert this state to a definition dictionary.

Returns:definition

class sfini.state.CanRetry(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter())

Bases: sfini.state._base.State

Retryable state mix-in.

Parameters:

• name – name of state
• comment – state description
• input_path – state input filter JSONPath, None for empty input
• output_path – state output filter JSONPath, None for discarded output

retriers

error handler policies

retry_for(errors: Sequence[str], interval: int = DefaultParameter(), max_attempts: int = DefaultParameter(), backoff_rate: float = DefaultParameter())

Add a retry handler.

Parameters:

• errors – codes of errors for retry to be executed. See AWS Step Functions documentation
• interval – (initial) retry interval (seconds)
• max_attempts – maximum number of attempts before re-raising error
• backoff_rate – retry interval increase factor between attempts

to_dict()

Convert this state to a definition dictionary.

Returns:definition

class sfini.state.CanCatch(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter())

Bases: sfini.state._base.State

Exception catching state mix-in.

Parameters:

• name – name of state
• comment – state description
• input_path – state input filter JSONPath, None for empty input
• output_path – state output filter JSONPath, None for discarded output

catchers

error handler policies

add_to(states)

Add this state to a state-machine definition.

Any child states will also be added to the definition.

Parameters:states (dict[str, State]) – state-machine states

catch(errors: Sequence[str], next_state: sfini.state._base.State, result_path: Optional[str] = DefaultParameter())

Add an error handler.

Parameters:

• errors – code of errors for catch clause to be executed. See AWS Step Functions documentation
• next_state – state to execute for catch clause
• result_path – error details location JSONPath

to_dict()

Convert this state to a definition dictionary.

Returns:definition

class sfini.state.Succeed(name: str, comment: str = DefaultParameter(), input_path: Optional[str] = DefaultParameter(), output_path: Optional[str] = DefaultParameter())

Bases: sfini.state._base.State

End execution successfully.

Parameters:

• name – name of state
• comment – state description
• input_path – state input filter JSONPath, None for empty input
• output_path – state output filter JSONPath, None for discarded output

class sfini.state.Fail(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter(), cause: str = DefaultParameter(), error: str = DefaultParameter())

Bases: sfini.state._base.State

End execution unsuccessfully.

Parameters:

• name – name of state
• comment – state description
• input_path – state input filter JSONPath, None for empty input
• output_path – state output filter JSONPath, None for discarded output
• error – error type
• cause – failure description

to_dict()

Convert this state to a definition dictionary.

Returns:definition

class sfini.state.Pass(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter(), result_path=DefaultParameter(), result: Union[None, bool, str, int, float, List[JSONable], Dict[str, JSONable]] = DefaultParameter())

Bases: sfini.state._base.HasResultPath, sfini.state._base.HasNext, sfini.state._base.State

No-op state, possibly introducing data.

The name specifies the location of any introduced data.

Parameters:

• name – name of state
• comment – state description
• input_path – state input filter JSONPath, None for empty input
• output_path – state output filter JSONPath, None for discarded output
• result_path – task output location JSONPath, None for discarded output
• result – return value of state, stored in the variable name

next

next state to execute

to_dict()

Convert this state to a definition dictionary.

Returns:definition

class sfini.state.Wait(name, until: Union[int, datetime.datetime, str], comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter())

Bases: sfini.state._base.HasNext, sfini.state._base.State

Wait for a time before continuing.

Parameters:

• name – name of state
• until – time to wait. If int, then seconds to wait; if datetime.datetime, then time to wait until; if str, then name of state-variable containing time to wait until
• comment – state description
• input_path – state input filter JSONPath, None for empty input
• output_path – state output filter JSONPath, None for discarded output

next

next state to execute

to_dict()

Convert this state to a definition dictionary.

Returns:definition

class sfini.state.Parallel(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter(), result_path=DefaultParameter())

Bases: sfini.state._base.HasResultPath, sfini.state._base.HasNext, sfini.state._base.CanRetry, sfini.state._base.CanCatch, sfini.state._base.State

Run states-machines in parallel.

Parameters:

• name – name of state
• comment – state description
• input_path – state input filter JSONPath, None for empty input
• output_path – state output filter JSONPath, None for discarded output
• result_path – task output location JSONPath, None for discarded output

branches

state-machines to run in parallel. These state-machines do not need to be registered with AWS Step Functions.

Type:list[sfini.StateMachine]

next

next state to execute

retriers

retry conditions

catchers

handled state errors

add(state_machine)

Add a state-machine to be executed.

The input to the state-machine execution is the input into this parallel state. The output of the parallel state is a list of each state-machine’s output (in order of adding).

Parameters:state_machine (sfini.StateMachine) – state-machine to add. It will be run when this task is executed. Added state-machines do not need to be registered with AWS Step Functions

to_dict()

Convert this state to a definition dictionary.

Returns:definition

class sfini.state.Choice(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter())

Bases: sfini.state._base.State

Branch execution based on comparisons.

Parameters:

• name – name of state
• comment – state description
• input_path – state input filter JSONPath, None for empty input
• output_path – state output filter JSONPath, None for discarded output

choices

choice rules determining branch conditions

Type:list[sfini.choice.ChoiceRule]

default

fall-back state if all comparisons fail, or None for no fall-back (Step Functions will raise a ‘States.NoChoiceMatched’ error)

add(rule)

Add a choice-rule.

Parameters:rule (sfini.choice.ChoiceRule) – branch execution condition and specification to add Raises:RuntimeError – rule doesn’t specify next-state

add_to(states)

Add this state to a state-machine definition.

Any child states will also be added to the definition.

Parameters:states (dict[str, State]) – state-machine states

remove(rule)

Remove a branch.

Parameters:rule (sfini.choice.ChoiceRule) – branch execution condition and specification to remove Raises:ValueError – if rule is not a registered branch

set_default(state: sfini.state._base.State)

Set the default state to execute when no conditions were met.

Parameters:state – default state to execute

to_dict()

Convert this state to a definition dictionary.

Returns:definition

class sfini.state.Task(name, resource, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter(), result_path=DefaultParameter(), timeout: int = DefaultParameter())

Bases: sfini.state._base.HasResultPath, sfini.state._base.HasNext, sfini.state._base.CanRetry, sfini.state._base.CanCatch, sfini.state._base.State

Activity execution.

Parameters:

• name – name of state
• resource (sfini.task_resource.TaskResource) – task executor, eg activity or Lambda function
• comment – state description
• input_path – state input filter JSONPath, None for empty input
• output_path – state output filter JSONPath, None for discarded output
• result_path – task output location JSONPath, None for discarded output
• timeout – seconds before task time-out

next

next state to execute

retriers

retry conditions

catchers

handled state errors

to_dict()

Convert this state to a definition dictionary.

Returns:definition