flowrep.nodes.atomic_recipe module

class flowrep.nodes.atomic_recipe.AtomicRecipe(*, type: Literal[RecipeElementType.ATOMIC] = RecipeElementType.ATOMIC, 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, reference: PythonReference, unpack_mode: UnpackMode = UnpackMode.TUPLE)

Bases: NodeRecipe

Atomos: uncuttable, indivisible.

A node representing a python function call.

Intended recipe realization: - Atomic nodes do not have internal structure from the perspective of a workflow

graph.

• The actions _inside_ them are ephemeral and not available for retrospective

  inspection.

• As with all nodes, their IO should be available for retrospective inspection.

• The conversion of function return values to node outputs is controlled via the

  unpack_mode flag.

type

The node type – always “atomic”.

Type:

Literal[base_models.RecipeElementType.ATOMIC]

inputs

The available input port names.

outputs

The available output port names.

reference

Info about the underlying python function.

Type:

base_models.PythonReference

unpack_mode

How to handle return values from running functions in atomic nodes.

Type:

UnpackMode

Properties:

fully_qualified_name: The fully qualified name of the function to call, i.e.

module and qualname as a dot-separated string.

check_outputs_when_not_unpacking()

property fully_qualified_name: str

property inputs_with_defaults: Annotated[list[Annotated[str, BeforeValidator(func=_validate_label, json_schema_input_type=PydanticUndefined)]], AfterValidator(func=validate_unique)]

model_config: ClassVar[ConfigDict] = {}

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

reference: base_models.PythonReference

type: Literal[base_models.RecipeElementType.ATOMIC]

unpack_mode: UnpackMode

class flowrep.nodes.atomic_recipe.UnpackMode(*values)

Bases: StrEnum

How to handle return values from running functions in atomic nodes.

• NONE: Return the output as a single value

• TUPLE: Split return into one port per tuple element

• DATACLASS: Split return into one port per dataclass field

DATACLASS = 'dataclass'

NONE = 'none'

TUPLE = 'tuple'