pyiron_workflow.dag module

class pyiron_workflow.dag.Macro(recipe: RecipeType, label: Annotated[str, BeforeValidator(func=_validate_label, json_schema_input_type=PydanticUndefined)] | None = None, /, **connections: Port | Node | JSONABLE)

Bases: ImmutableDag

evaluate(run: Run[ResultType], config: RunConfig) → Run[ResultType]

property function_metadata: FunctionMetadata | None

validate(do_types: bool = True, do_ontology: bool = True, extra_knowledge: rdflib.Graph | None = None) → validation.CombinedValidationReport

Validate this node’s types and (optionally) ontology.

Thin wrapper around validation.validate_plan().

pyiron_workflow.dag.evaluate_dag_by_layer(nodes: NodeMap, run: Run[CompositeData], config: RunConfig) → None

pyiron_workflow.dag.evaluate_node(node: Node[Any, ResultType], label_in_run: Annotated[str, BeforeValidator(func=_validate_label, json_schema_input_type=PydanticUndefined)], run: Run[CompositeData], config: RunConfig)

pyiron_workflow.dag.gather_target_inputs(node_label: Annotated[str, BeforeValidator(func=_validate_label, json_schema_input_type=PydanticUndefined)], runtime_data: CompositeData) → dict[str, Any]

Resolve input values for a target node from graph input ports and sibling output ports according to the graph recipe edges.

Ports not covered by any edge are omitted — the child’s own defaults (if any) will be used downstream.

pyiron_workflow.dag.populate_outputs(result: CompositeData) → None

pyiron_workflow.dag.topo_sort_nodes(nodes: NodeMap, edges: dict[TargetHandle, SourceHandle]) → list[list[Annotated[str, BeforeValidator(func=_validate_label, json_schema_input_type=PydanticUndefined)]]]

Kahn’s algorithm over sibling edges, grouped into independent layers.

Each layer contains nodes whose dependencies all live in earlier layers, so members of a layer may be executed concurrently. Deterministic tie-breaking by label within each layer.