sfini.state¶
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())[source]¶ 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
-
class
sfini.state.
HasNext
(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter())[source]¶ 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)[source]¶ 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
-
class
sfini.state.
HasResultPath
(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter(), result_path: Optional[str] = DefaultParameter())[source]¶ 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
-
class
sfini.state.
CanRetry
(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter())[source]¶ 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())[source]¶ 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
-
class
sfini.state.
CanCatch
(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter())[source]¶ 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)[source]¶ 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())[source]¶ 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
-
class
sfini.state.
Succeed
(name: str, comment: str = DefaultParameter(), input_path: Optional[str] = DefaultParameter(), output_path: Optional[str] = DefaultParameter())[source]¶ 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())[source]¶ 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
-
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())[source]¶ 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
-
class
sfini.state.
Wait
(name, until: Union[int, datetime.datetime, str], comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter())[source]¶ 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; ifdatetime.datetime
, then time to wait until; ifstr
, 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
-
class
sfini.state.
Parallel
(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter(), result_path=DefaultParameter())[source]¶ 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)[source]¶ 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
-
class
sfini.state.
Choice
(name, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter())[source]¶ 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)[source]¶ 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)[source]¶ 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)[source]¶ Remove a branch.
Parameters: rule (sfini.choice.ChoiceRule) – branch execution condition and specification to remove Raises: ValueError
– if rule is not a registered branch
-
class
sfini.state.
Task
(name, resource, comment=DefaultParameter(), input_path=DefaultParameter(), output_path=DefaultParameter(), result_path=DefaultParameter(), timeout: int = DefaultParameter())[source]¶ 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