flowrep.nodes.while_recipe module

class flowrep.nodes.while_recipe.WhileRecipe(*, type: Literal[RecipeElementType.WHILE] = RecipeElementType.WHILE, inputs: Annotated[list[Annotated[str, BeforeValidator(func=_validate_label, json_schema_input_type=PydanticUndefined)]], AfterValidator(func=validate_unique)], outputs: Annotated[list[Annotated[str, BeforeValidator(func=_validate_label, json_schema_input_type=PydanticUndefined)]], AfterValidator(func=validate_unique)], description: str | None = None, case: ConditionalCase, input_edges: dict[TargetHandle, InputSource], output_edges: dict[OutputTarget, SourceHandle | InputSource])

Bases: NodeRecipe

A loop node that repeatedly executes a body while a condition is true. This is a dynamic node, which must actualize the body of its subgraph at runtime. Output labels must be a subset of input labels to facilitate unambiguous looping and guarantee output availability even if the body never executes. Output edges _must_ come from the case’s body node, not its condition node. It is the responsibility of the WfMS to repeat execution, inferring from the outputs (and their correspondence with input labels) which data needs to looped on such that body instance outputs get passed to the next set of condition and body instances. It is also the responsibility of the WfMS to leverage loop inputs as fallbacks in case the body never executes, in this way we are a little cheeky about the recipe having static outputs – the actual sourcing is runtime dependent, but you are guaranteed to have data there at the end of the day.

Intended recipe realization:

1. The case condition recipe is used to instantiate a condition node

2. Input edges are routed to the condition

3. The condition is executed and evaluated

   1. The evaluation port is specified in the case

4. If the condition evaluates False, terminate; route while-loop inputs to

   outputs of matching label to guarantee data availability.

5. Else, the case body recipe is used to instantiate a body node

6. Input edges are routed to the body, and it is executed

7. Another condition node is instantiated, and the correspondence of loop

   output labels and input labels is leveraged to infer what inputs need to come from the loop input, and what from the most recent body node instance

8. If the condition evaluates False, terminate and route the most recent

   body node output to the loop output

9. Else, instantiate another body node and follow the same output-input label

   comparison to infer which input data is still coming from the loop vs. which is coming from the last body node instance

10. Repeat steps 7-9 until the condition evaluates False.

type

The node type – always “while”.

Type:

Literal[flowrep.base_models.RecipeElementType.WHILE]

inputs

The available input port names.

Type:

list[str]

outputs

The available output port names. For while-nodes these _must_ be a subset of the input port names.

Type:

list[str]

case

The condition-body pair to be looped over by repeated instantiation. The condition node must produce a boolean output (specified by condition_output or inferred if the condition has exactly one output).

Type:

flowrep.nodes.helper_models.ConditionalCase

input_edges

Edges from workflow inputs to the initial condition/body nodes. Keys are targets on condition/body, values are workflow input ports.

Type:

dict[flowrep.edge_models.TargetHandle, flowrep.edge_models.InputSource]

output_edges

Edges from the body of the conditional case to the outputs. The actual runtime output edges are dependent on whether the condition ever evaluated to be true, so these output edges are constrained (no coming from the condition node) and pseudo-prospective (pass-through input may be leveraged at runtime in a fixed way).

Type:

dict[flowrep.edge_models.OutputTarget, flowrep.edge_models.SourceHandle | flowrep.edge_models.InputSource]

property body_body_edges: dict[TargetHandle, SourceHandle]

Inferred edges for passing body output to next body iteration.

property body_condition_edges: dict[TargetHandle, SourceHandle]

Inferred edges for passing body output to next condition evaluation.

case: helper_models.ConditionalCase

input_edges: edge_models.InputEdges

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

output_edges: edge_models.OutputEdges

property prospective_nodes: Recipes

type: Literal[base_models.RecipeElementType.WHILE]

validate_internal_data_completeness()

validate_io_edges()

validate_output_is_subset_of_inputs()