aiida.orm.nodes.process.workflow package#

Module with Node sub classes for workflow processes.

Submodules#

Module with Node sub class for workchain processes.

class aiida.orm.nodes.process.workflow.workchain.WorkChainNode(backend: StorageBackend | None = None, user: User | None = None, computer: Computer | None = None, **kwargs: Any)[源代码]#

基类：WorkflowNode

ORM class for all nodes representing the execution of a WorkChain.

STEPPER_STATE_INFO_KEY = 'stepper_state_info'#

__abstractmethods__ = frozenset({})#

__module__ = 'aiida.orm.nodes.process.workflow.workchain'#

__parameters__ = ()#

__plugin_type_string: ClassVar[str]#

__query_type_string: ClassVar[str]#

_abc_impl = <_abc._abc_data object>#

_logger: Logger | None = <Logger aiida.orm.nodes.process.workflow.workchain.WorkChainNode (WARNING)>#

_updatable_attributes: Tuple[str, ...] = ('sealed', 'paused', 'checkpoints', 'exception', 'exit_message', 'exit_status', 'process_label', 'process_state', 'process_status', 'stepper_state_info')#

fields: QbFields = {'attributes': 'QbDictField(attributes.*) -> Dict[str, Any]', 'computer_pk': 'QbNumericField(attributes.computer_pk) -> Optional[int]', 'ctime': 'QbNumericField(ctime) -> datetime', 'description': 'QbStrField(description) -> str', 'exception': 'QbStrField(attributes.exception) -> Optional[str]', 'exit_message': 'QbStrField(attributes.exit_message) -> Optional[str]', 'exit_status': 'QbNumericField(attributes.exit_status) -> Optional[int]', 'extras': 'QbDictField(extras.*) -> Dict[str, Any]', 'label': 'QbStrField(label) -> str', 'mtime': 'QbNumericField(mtime) -> datetime', 'node_type': 'QbStrField(node_type) -> str', 'paused': 'QbField(attributes.paused) -> bool', 'pk': 'QbNumericField(pk) -> int', 'process_label': 'QbStrField(attributes.process_label) -> Optional[str]', 'process_state': 'QbStrField(attributes.process_state) -> Optional[str]', 'process_status': 'QbStrField(attributes.process_status) -> Optional[str]', 'process_type': 'QbStrField(attributes.process_type) -> Optional[str]', 'repository_metadata': 'QbDictField(repository_metadata) -> Dict[str, Any]', 'sealed': 'QbField(attributes.sealed) -> bool', 'user_pk': 'QbNumericField(user_pk) -> int', 'uuid': 'QbStrField(uuid) -> str'}#

set_stepper_state_info(stepper_state_info: str) → None[源代码]#

Set the stepper state info

参数:: state – string representation of the stepper state info

property stepper_state_info: str | None#

Return the stepper state info

返回:: string representation of the stepper state info

Module with Node sub class for workflow processes.

class aiida.orm.nodes.process.workflow.workflow.WorkflowNode(backend: StorageBackend | None = None, user: User | None = None, computer: Computer | None = None, **kwargs: Any)[源代码]#

基类：ProcessNode

Base class for all nodes representing the execution of a workflow process.

_CLS_NODE_LINKS#: WorkflowNodeLinks 的别名

__abstractmethods__ = frozenset({})#

__module__ = 'aiida.orm.nodes.process.workflow.workflow'#

__parameters__ = ()#

__plugin_type_string: ClassVar[str]#

__query_type_string: ClassVar[str]#

_abc_impl = <_abc._abc_data object>#

_logger: Logger | None = <Logger aiida.orm.nodes.process.workflow.workflow.WorkflowNode (WARNING)>#

_storable = True#

_unstorable_message = 'storing for this node has been disabled'#

fields: QbFields = {'attributes': 'QbDictField(attributes.*) -> Dict[str, Any]', 'computer_pk': 'QbNumericField(attributes.computer_pk) -> Optional[int]', 'ctime': 'QbNumericField(ctime) -> datetime', 'description': 'QbStrField(description) -> str', 'exception': 'QbStrField(attributes.exception) -> Optional[str]', 'exit_message': 'QbStrField(attributes.exit_message) -> Optional[str]', 'exit_status': 'QbNumericField(attributes.exit_status) -> Optional[int]', 'extras': 'QbDictField(extras.*) -> Dict[str, Any]', 'label': 'QbStrField(label) -> str', 'mtime': 'QbNumericField(mtime) -> datetime', 'node_type': 'QbStrField(node_type) -> str', 'paused': 'QbField(attributes.paused) -> bool', 'pk': 'QbNumericField(pk) -> int', 'process_label': 'QbStrField(attributes.process_label) -> Optional[str]', 'process_state': 'QbStrField(attributes.process_state) -> Optional[str]', 'process_status': 'QbStrField(attributes.process_status) -> Optional[str]', 'process_type': 'QbStrField(attributes.process_type) -> Optional[str]', 'repository_metadata': 'QbDictField(repository_metadata) -> Dict[str, Any]', 'sealed': 'QbField(attributes.sealed) -> bool', 'user_pk': 'QbNumericField(user_pk) -> int', 'uuid': 'QbStrField(uuid) -> str'}#

property inputs: NodeLinksManager#

Return an instance of NodeLinksManager to manage incoming INPUT_WORK links

The returned Manager allows you to easily explore the nodes connected to this node via an incoming INPUT_WORK link. The incoming nodes are reachable by their link labels which are attributes of the manager.

返回:: NodeLinksManager

property outputs: NodeLinksManager#

Return an instance of NodeLinksManager to manage outgoing RETURN links

The returned Manager allows you to easily explore the nodes connected to this node via an outgoing RETURN link. The outgoing nodes are reachable by their link labels which are attributes of the manager.

返回:: NodeLinksManager

class aiida.orm.nodes.process.workflow.workflow.WorkflowNodeLinks(node: Node)[源代码]#

基类：ProcessNodeLinks

Interface for links of a node instance.

__annotations__ = {}#

__module__ = 'aiida.orm.nodes.process.workflow.workflow'#

validate_outgoing(target: Node, link_type: LinkType, link_label: str) → None[源代码]#

Validate adding a link of the given type from ourself to a given node.

A workflow cannot ‘create’ Data, so if we receive an outgoing link to an unstored Data node, that means the user created a Data node within our function body and tries to attach it as an output. This is strictly forbidden and can cause provenance to be lost.

参数:

target – the node to which the link is going
link_type – the link type
link_label – the link label

抛出:

TypeError – if target is not a Node instance or link_type is not a LinkType enum
ValueError – if the proposed link is invalid

Module with Node sub class for workflow function processes.

class aiida.orm.nodes.process.workflow.workfunction.WorkFunctionNode(backend: StorageBackend | None = None, user: User | None = None, computer: Computer | None = None, **kwargs: Any)[源代码]#

基类：FunctionCalculationMixin, WorkflowNode

ORM class for all nodes representing the execution of a workfunction.

_CLS_NODE_LINKS#: WorkFunctionNodeLinks 的别名

__abstractmethods__ = frozenset({})#

__module__ = 'aiida.orm.nodes.process.workflow.workfunction'#

__parameters__ = ()#

_abc_impl = <_abc._abc_data object>#

_logger: Logger | None = <Logger aiida.orm.nodes.process.workflow.workfunction.WorkFunctionNode (WARNING)>#

fields: QbFields = {'attributes': 'QbDictField(attributes.*) -> Dict[str, Any]', 'computer_pk': 'QbNumericField(attributes.computer_pk) -> Optional[int]', 'ctime': 'QbNumericField(ctime) -> datetime', 'description': 'QbStrField(description) -> str', 'exception': 'QbStrField(attributes.exception) -> Optional[str]', 'exit_message': 'QbStrField(attributes.exit_message) -> Optional[str]', 'exit_status': 'QbNumericField(attributes.exit_status) -> Optional[int]', 'extras': 'QbDictField(extras.*) -> Dict[str, Any]', 'label': 'QbStrField(label) -> str', 'mtime': 'QbNumericField(mtime) -> datetime', 'node_type': 'QbStrField(node_type) -> str', 'paused': 'QbField(attributes.paused) -> bool', 'pk': 'QbNumericField(pk) -> int', 'process_label': 'QbStrField(attributes.process_label) -> Optional[str]', 'process_state': 'QbStrField(attributes.process_state) -> Optional[str]', 'process_status': 'QbStrField(attributes.process_status) -> Optional[str]', 'process_type': 'QbStrField(attributes.process_type) -> Optional[str]', 'repository_metadata': 'QbDictField(repository_metadata) -> Dict[str, Any]', 'sealed': 'QbField(attributes.sealed) -> bool', 'user_pk': 'QbNumericField(user_pk) -> int', 'uuid': 'QbStrField(uuid) -> str'}#

class aiida.orm.nodes.process.workflow.workfunction.WorkFunctionNodeLinks(node: Node)[源代码]#

基类：WorkflowNodeLinks

Interface for links of a node instance.

__annotations__ = {}#

__module__ = 'aiida.orm.nodes.process.workflow.workfunction'#

validate_outgoing(target: Node, link_type: LinkType, link_label: str) → None[源代码]#

Validate adding a link of the given type from ourself to a given node.

A workfunction cannot create Data, so if we receive an outgoing RETURN link to an unstored Data node, that means the user created a Data node within our function body and is trying to return it. This use case should be reserved for @calcfunctions, as they can have CREATE links.

参数:

target – the node to which the link is going
link_type – the link type
link_label – the link label

抛出:

TypeError – if target is not a Node instance or link_type is not a LinkType enum
ValueError – if the proposed link is invalid