plumpy.workchains module

plumpy.workchains.ToContext

alias of dict

class plumpy.workchains.WorkChain(*args: Any, **kwargs: Any)

Bases: ContextMixin, Process

A WorkChain is a series of instructions carried out with the ability to save state in between.

_CONTEXT = 'CONTEXT'

_STEPPER_STATE = 'stepper_state'

__called: bool

_abc_impl = <_abc._abc_data object>

_awaitables: Dict[asyncio.Future | processes.Process, str]

_context: AttributesDict | None

_creation_time: float | None

_do_step() → Any

_event_callbacks: Dict[Hashable, List[EVENT_CALLBACK_TYPE]]

_outputs: Dict[str, Any]

_parsed_inputs: utils.AttributesFrozendict | None

_pre_paused_status: str | None

_spec_class

alias of WorkChainSpec

_state: State | None

_status: str | None

_stepper: Stepper | None

_uuid: uuid.UUID | None

classmethod get_state_classes() → Dict[Hashable, Type[State]]

load_instance_state(saved_state: MutableMapping[str, Any], load_context: LoadSaveContext) → None

Load the process from its saved instance state.

Parameters:

• saved_state – A bundle to load the state from

• load_context – The load context

on_create() → None

async run() → Any

This function will be run when the process is triggered. It should be overridden by a subclass.

save_instance_state(out_state: MutableMapping[str, Any], save_context: LoadSaveContext | None) → None

Add the instance state to out_state. .. important:

The instance state will contain a pointer to the ``ctx``,
and so should be deep copied or serialised before persisting.

classmethod spec() → WorkChainSpec

to_context(**kwargs: Future | Process) → None

This is a convenience method that provides syntactic sugar, for a user to add multiple intersteps that will assign a certain value to the corresponding key in the context of the workchain

class plumpy.workchains.WorkChainSpec

Bases: ProcessSpec

get_description() → Dict[str, str]

Get a description of this process specification

Returns:

a dictionary with the descriptions of the input and output port namespaces

get_outline() → _Instruction | _FunctionCall

outline(*commands: _Instruction | Callable[[WorkChain], Any]) → None

Define the outline that describes this work chain.

Parameters:

commands – One or more functions that make up this work chain.

plumpy.workchains.if_(condition: Callable[[WorkChain], bool]) → _If

A conditional that can be used in a workchain outline.

Use as:

if_(cls.conditional)(
  cls.step1,
  cls.step2
)

Each step can, of course, also be any valid workchain step e.g. conditional.

Parameters:

condition – The workchain method that will return True or False

plumpy.workchains.return_ = <plumpy.workchains._Return object>

A global singleton that contains a Return instruction that allows to exit out of the workchain outline directly with None as exit code To set a specific exit code, call it with the desired exit code

Use as:

if_(cls.conditional)(
  return_
)

or:

if_(cls.conditional)(
  return_(EXIT_CODE)
)

Parameters:

exit_code – an integer exit code to pass as the return value, None by default

plumpy.workchains.while_(condition: Callable[[WorkChain], bool]) → _While

A while loop that can be used in a workchain outline.

Use as:

while_(cls.conditional)(
  cls.step1,
  cls.step2
)

Each step can, of course, also be any valid workchain step e.g. conditional.

Parameters:

condition – The workchain method that will return True or False