sfini.execution.history¶
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)[source]¶ 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)[source]¶ 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)[source]¶ 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[source]¶ 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)[source]¶ 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)[source]¶ 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)[source]¶ 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())[source]¶ 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())[source]¶ 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())[source]¶ 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][source]¶ List the execution history.
Parameters: history_events – history events as provided by AWS API Returns: history of execution events