plumpy.base package

plumpy.base package#

class plumpy.base.StateMachine(*args: Any, **kwargs: Any)[source]#

Bases: object

STATES: Sequence[Type[State]] | None = None#
_STATES_MAP: Dict[Hashable, Type[State]] | None = None#
classmethod __ensure_built() → None#
_create_state_instance(state_cls: Hashable, **kwargs: Any) → State[source]#
_enter_next_state(next_state: State) → None[source]#
_exit_current_state(next_state: State) → None[source]#

Exit the given state

_fire_state_event(hook: Hashable, state: State | None) → None[source]#
_transition_failing = False#
_transitioning = False#
add_state_event_callback(hook: Hashable, callback: Callable[[StateMachine, Hashable, State | None], None]) → None[source]#

Add a callback to be called on a particular state event hook. The callback should have form fn(state_machine, hook, state)

Parameters:

  • hook – The state event hook

  • callback – The callback function

create_initial_state() → State[source]#
create_state(state_label: Hashable, *args: Any, **kwargs: Any) → State[source]#
get_debug() → bool[source]#
classmethod get_state_class(label: None | Enum | str) → Type[State][source]#
classmethod get_states() → Sequence[Type[State]][source]#
classmethod get_states_map() → Dict[Hashable, Type[State]][source]#
init(*args: Any, **kwargs: Any) → None#
classmethod initial_state_label() → None | Enum | str[source]#
on_terminated(*args: Any, **kwargs: Any) → None#
remove_state_event_callback(hook: Hashable, callback: Callable[[StateMachine, Hashable, State | None], None]) → None[source]#
set_debug(enabled: bool) → None[source]#
property state: None | Enum | str#
transition_failed(initial_state: Hashable, final_state: Hashable, exception: Exception, trace: TracebackType) → None[source]#

Called when a state transitions fails.

This method can be overwritten to change the default behaviour which is to raise the exception.

Parameters:

exception – The transition failed exception.

transition_to(new_state: State | None, **kwargs: Any) → None[source]#

Transite to the new state.

The new target state will be create lazily when the state is not yet instantiated, which will happened for states not in the expect path such as pause and kill. The arguments are passed to the state class to create state instance. (process arg does not need to pass since it will always call with ‘self’ as process)

class plumpy.base.StateMachineMeta[source]#

Bases: type

exception plumpy.base.TransitionFailed(initial_state: State, final_state: State | None = None, traceback_str: str | None = None)[source]#

Bases: Exception

A state transition failed

_format_msg() → str[source]#
plumpy.base.call_with_super_check(wrapped: Callable[[...], Any], *args: Any, **kwargs: Any) → None[source]#

Call a class method checking that all subclasses called super along the way

plumpy.base.event(from_states: str | Type[State] | Iterable[Type[State]] = '*', to_states: str | Type[State] | Iterable[Type[State]] = '*') → Callable[[Callable[[...], Any]], Callable[[...], Any]][source]#

A decorator to check for correct transitions, raising EventError on invalid transitions.

plumpy.base.super_check(wrapped: Callable[[...], Any]) → Callable[[...], Any][source]#

Decorator to add a super check to a function to be used with call_with_super_check

Submodules#

Contents