flowrep.nodes.try_recipe module

class flowrep.nodes.try_recipe.TryRecipe(*, type: Literal[RecipeElementType.TRY] = RecipeElementType.TRY, 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, try_node: LabeledRecipe, exception_cases: list[ExceptionCase], input_edges: dict[TargetHandle, InputSource], prospective_output_edges: dict[OutputTarget, Annotated[list[SourceHandle], AfterValidator(func=validate_unique)]])

Bases: NodeRecipe

Try and except your way through a series of exceptions, with the option to perform a finally step. This is a dynamic node, which must actualize the body of its subgraph at runtime.

Intended recipe realization: 1. Instantiate the try node 2. Connect input to this node according to input edges 3. Execute and evaluate the try node 4. If an exception is encountered, walk through the exception cases 5. If an exception match is found encountered, repeat (1-3) for the case body 7. Use the matrix of output edges to connect the greatest possible extent of

output to the outputs of the successful try/except case (if any)

type

The node type – always “try”.

Type:

Literal[flowrep.base_models.RecipeElementType.TRY]

inputs

The available input port names.

Type:

list[str]

outputs

The available output port names.

Type:

list[str]

try_node

The primary body node to execute.

Type:

flowrep.nodes.helper_models.LabeledRecipe

exception_cases

The exception type-body pairs to be walked over searching for an exception match in the event that the try node fails.

Type:

list[flowrep.nodes.helper_models.ExceptionCase]

input_edges

Edges from workflow inputs to inputs of the prospective nodes.

Type:

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

prospective_output_edges

For each output, sources from possible try/except nodes. Note that at most one of these possible edges will be actualized at runtime based on which try/except case node actually runs without exception (if any).

Type:

dict[flowrep.edge_models.OutputTarget, list[flowrep.edge_models.SourceHandle]]

Note

While each available output must be represented in the prospective_output_edges, not every possible try/except branch needs to be represented in the rows of this matrix. It is thus possible that some outputs are left with no source and thus non-data values at the end of the node’s execution.

exception_cases: list[helper_models.ExceptionCase]

input_edges: edge_models.InputEdges

model_config: ClassVar[ConfigDict] = {}

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

property prospective_nodes: Recipes

prospective_output_edges: dict[edge_models.OutputTarget, base_models.UniqueList[edge_models.SourceHandle]]

try_node: helper_models.LabeledRecipe

type: Literal[base_models.RecipeElementType.TRY]

classmethod validate_exception_cases_not_empty(v)

validate_internal_data_completeness()

validate_io_edges()

validate_prospective_nodes_have_unique_labels()