plumpy.base package

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

Bases: object

STATES: Optional[Sequence[Type[plumpy.base.state_machine.State]]] = None
_STATES_MAP: Optional[Dict[Hashable, Type[plumpy.base.state_machine.State]]] = None
classmethod __ensure_built()None
_create_state_instance(state: Union[Hashable, plumpy.base.state_machine.State, Type[plumpy.base.state_machine.State]], *args: Any, **kwargs: Any) → plumpy.base.state_machine.State[source]
_ensure_state_class(state: Union[Hashable, Type[plumpy.base.state_machine.State]]) → Type[plumpy.base.state_machine.State][source]
_enter_next_state(next_state: plumpy.base.state_machine.State)None[source]
_exit_current_state(next_state: plumpy.base.state_machine.State)None[source]

Exit the given state

_fire_state_event(hook: Hashable, state: Optional[plumpy.base.state_machine.State])None[source]
_transition_failing = False
_transitioning = False
add_state_event_callback(hook: Hashable, callback: Callable[[StateMachine, Hashable, Optional[State]], 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() → plumpy.base.state_machine.State[source]
create_state(state_label: Hashable, *args: Any, **kwargs: Any) → plumpy.base.state_machine.State[source]
get_debug()bool[source]
classmethod get_state_class(label: Union[None, enum.Enum, str]) → Type[plumpy.base.state_machine.State][source]
classmethod get_states() → Sequence[Type[plumpy.base.state_machine.State]][source]
classmethod get_states_map() → Dict[Hashable, Type[plumpy.base.state_machine.State]][source]
init(*args: Any, **kwargs: Any)None
classmethod initial_state_label() → Union[None, enum.Enum, str][source]
on_terminated(*args: Any, **kwargs: Any)None
remove_state_event_callback(hook: Hashable, callback: Callable[[StateMachine, Hashable, Optional[State]], None])None[source]
set_debug(enabled: bool)None[source]
property state
static transition_failed(initial_state: Hashable, final_state: Hashable, exception: Exception, trace: traceback)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: Union[Hashable, plumpy.base.state_machine.State, Type[plumpy.base.state_machine.State]], *args: Any, **kwargs: Any)None[source]
class plumpy.base.StateMachineMeta[source]

Bases: type

exception plumpy.base.TransitionFailed(initial_state: plumpy.base.state_machine.State, final_state: Optional[State] = None, traceback_str: Optional[str] = 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: Union[str, Type[State], Iterable[Type[State]]] = '*', to_states: Union[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