sfini.execution.history module

State-machine execution history events.

Use sfini.execution.Execution.format_history for nice history printing.

class sfini.execution.history.ActivityScheduled(timestamp, event_type, event_id, resource, previous_event_id=None, task_input=DefaultParameter(), timeout=None, heartbeat: int = None)

Bases: sfini.execution.history.LambdaFunctionScheduled

An execution history activity task-schedule event.

Parameters:

• timestamp – event time-stamp
• event_type – type of event
• event_id – identifying index of event
• resource – AWS Lambda function ARN
• previous_event_id – identifying index of causal event
• task_input – task input
• timeout – time-out (seconds) of task execution
• heartbeat – heartbeat time-out (seconds)

class sfini.execution.history.ActivityStarted(timestamp, event_type, event_id, worker_name: str, previous_event_id=None)

Bases: sfini.execution.history.Event

An execution history activity task-start event.

Parameters:

• timestamp – event time-stamp
• event_type – type of event
• event_id – identifying index of event
• worker_name – name of activity worker executing activity task
• previous_event_id – identifying index of causal event

details_str

Format the event details.

Returns:event details, formatted as string Return type:str

class sfini.execution.history.Event(timestamp, event_type: str, event_id: int, previous_event_id: int = None)

Bases: object

An execution history event.

Parameters:

• timestamp (datetime.datetime) – event time-stamp
• event_type – type of event
• event_id – identifying index of event
• previous_event_id – identifying index of causal event

details_str

Format the event details.

Returns:event details, formatted as string Return type:str

classmethod from_history_event(history_event: Dict[str, Union[None, bool, str, int, float, List[JSONable], Dict[str, JSONable]]]) → Event

Parse an history event.

Parameters:history_event – execution history event date, provided by AWS API Returns:constructed execution history event Return type:Event

class sfini.execution.history.ExecutionStarted(timestamp, event_type, event_id, previous_event_id=None, execution_input: Union[None, bool, str, int, float, List[JSONable], Dict[str, JSONable]] = DefaultParameter(), role_arn: str = None)

Bases: sfini.execution.history.Event

An execution history execution-start event.

Parameters:

• timestamp – event time-stamp
• event_type – type of event
• event_id – identifying index of event
• previous_event_id – identifying index of causal event
• execution_input – execution input
• role_arn – execution AWS IAM role ARN

class sfini.execution.history.Failed(timestamp, event_type, event_id, previous_event_id=None, error: str = None, cause: str = None)

Bases: sfini.execution.history.Event

An execution history failure event.

Parameters:

• timestamp – event time-stamp
• event_type – type of event
• event_id – identifying index of event
• previous_event_id – identifying index of causal event
• error – error type
• cause – failure details

details_str

Format the event details.

Returns:event details, formatted as string Return type:str

class sfini.execution.history.LambdaFunctionScheduled(timestamp, event_type, event_id, resource: str, previous_event_id=None, task_input: Union[None, bool, str, int, float, List[JSONable], Dict[str, JSONable]] = DefaultParameter(), timeout: int = None)

Bases: sfini.execution.history.Event

An execution history AWS Lambda task-schedule event.

Parameters:

• timestamp – event time-stamp
• event_type – type of event
• event_id – identifying index of event
• resource – AWS Lambda function ARN
• previous_event_id – identifying index of causal event
• task_input – task input
• timeout – time-out (seconds) of task execution

details_str

Format the event details.

Returns:event details, formatted as string Return type:str

class sfini.execution.history.ObjectSucceeded(timestamp, event_type, event_id, previous_event_id=None, output: Union[None, bool, str, int, float, List[JSONable], Dict[str, JSONable]] = DefaultParameter())

Bases: sfini.execution.history.Event

An execution history succeed event.

Parameters:

• timestamp – event time-stamp
• event_type – type of event
• event_id – identifying index of event
• previous_event_id – identifying index of causal event
• output – output of state/execution

class sfini.execution.history.StateEntered(timestamp, event_type, event_id, state_name: str, previous_event_id=None, state_input: Union[None, bool, str, int, float, List[JSONable], Dict[str, JSONable]] = DefaultParameter())

Bases: sfini.execution.history.Event

An execution history state-enter event.

Parameters:

• timestamp – event time-stamp
• event_type – type of event
• event_id – identifying index of event
• state_name – state name
• previous_event_id – identifying index of causal event
• state_input – state input

details_str

Format the event details.

Returns:event details, formatted as string Return type:str

class sfini.execution.history.StateExited(timestamp, event_type, event_id, state_name: str, previous_event_id=None, output: Union[None, bool, str, int, float, List[JSONable], Dict[str, JSONable]] = DefaultParameter())

Bases: sfini.execution.history.Event

An execution history state-exit event.

Parameters:

• timestamp – event time-stamp
• event_type – type of event
• event_id – identifying index of event
• state_name – state name
• previous_event_id – identifying index of causal event
• output – state output

details_str

Format the event details.

Returns:event details, formatted as string Return type:str

sfini.execution.history.parse_history(history_events: List[Dict[str, Union[None, bool, str, int, float, List[JSONable], Dict[str, JSONable]]]]) → List[sfini.execution.history.Event]

List the execution history.

Parameters:history_events – history events as provided by AWS API Returns:history of execution events