Callable recipes

src/flowrep/base_models.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import abc
 import inspect
 import keyword
 from collections.abc import Hashable
@@ -94,7 +95,7 @@ def ensure_class_signature_from_init(cls: type, context: str) -> None:
 Labels = UniqueList[Label]
 
 
-class NodeRecipe(pydantic.BaseModel):
+class NodeRecipe(pydantic.BaseModel, abc.ABC):
     type: RecipeElementType
     inputs: Labels
     outputs: Labels
@@ -131,6 +132,7 @@ def _check_inputs_with_defaults_subset_of_inputs(self) -> Self:
     def validate_internal_data_completeness(self):
         return self
 
+    @abc.abstractmethod
     def __call__(self, *args, **kwargs):
         raise NotImplementedError(
             f"{self.__class__.__name__} is not a callable recipe type"

src/flowrep/prospective/for_recipe.py

@@ -179,3 +179,9 @@ def validate_internal_data_completeness(self):
             self.prospective_nodes, self.input_edges
         )
         return self
+
+    def __call__(self, *args, **kwargs):
+        from flowrep import wfms
+
+        data = wfms._run_for(self, **wfms.variadic_to_inputs(self, *args, **kwargs))
+        return wfms.data_to_return(data)

src/flowrep/prospective/if_recipe.py

@@ -116,3 +116,9 @@ def validate_internal_data_completeness(self):
             self.prospective_nodes, self.input_edges
         )
         return self
+
+    def __call__(self, *args, **kwargs):
+        from flowrep import wfms
+
+        data = wfms._run_if(self, **wfms.variadic_to_inputs(self, *args, **kwargs))
+        return wfms.data_to_return(data)

src/flowrep/prospective/try_recipe.py

@@ -103,3 +103,9 @@ def validate_internal_data_completeness(self):
             self.prospective_nodes, self.input_edges
         )
         return self
+
+    def __call__(self, *args, **kwargs):
+        from flowrep import wfms
+
+        data = wfms._run_try(self, **wfms.variadic_to_inputs(self, *args, **kwargs))
+        return wfms.data_to_return(data)

src/flowrep/prospective/while_recipe.py

@@ -143,3 +143,9 @@ def validate_internal_data_completeness(self):
             self.prospective_nodes, self.input_edges
         )
         return self
+
+    def __call__(self, *args, **kwargs):
+        from flowrep import wfms
+
+        data = wfms._run_while(self, **wfms.variadic_to_inputs(self, *args, **kwargs))
+        return wfms.data_to_return(data)

src/flowrep/prospective/workflow_recipe.py

@@ -87,10 +87,12 @@ def validate_internal_data_completeness(self):
 
     def __call__(self, *args, **kwargs):
         if self.reference is None:
-            raise ValueError(
-                f"{self.__class__.__name__} recipes are only callable when they are "
-                f"attached to an underlying python definiton in their reference field."
+            from flowrep import wfms
+
+            data = wfms._run_workflow(
+                self, **wfms.variadic_to_inputs(self, *args, **kwargs)
             )
+            return wfms.data_to_return(data)
         func = retrieve.import_from_string(self.reference.info.fully_qualified_name)
         return func(*args, **kwargs)
 

src/flowrep/wfms.py

@@ -30,14 +30,23 @@
 from flowrep.retrospective import datastructures
 
 
+def _unsupported_recipe(recipe: Any) -> TypeError:
+    return TypeError(f"Unsupported recipe type: {type(recipe).__name__}")
+
+
 def run_recipe(
     recipe: union_types.RecipeDiscrimination, **kwargs: Any
 ) -> datastructures.NodeData:
     """
     Execute a flowrep recipe, returning a populated :class:`LiveNode`.
 
     All inputs are passed as keyword arguments matching the recipe's input port names.
+    Inputs backed by a python default may be omitted; anything else must be supplied.
     """
+    if not isinstance(recipe, base_models.NodeRecipe):
+        # Guard before binding, which needs the recipe's input labels
+        raise _unsupported_recipe(recipe)
+    kwargs = variadic_to_inputs(recipe, **kwargs)
     match recipe:
         case atomic_recipe.AtomicRecipe():
             return _run_atomic(recipe, **kwargs)
@@ -54,7 +63,7 @@ def run_recipe(
         case while_recipe.WhileRecipe():
             return _run_while(recipe, **kwargs)
         case _:
-            raise TypeError(f"Unsupported recipe type: {type(recipe).__name__}")
+            raise _unsupported_recipe(recipe)
 
 
 # ---------------------------------------------------------------------------
@@ -205,6 +214,25 @@ def _populate_workflow_outputs(
 # ---------------------------------------------------------------------------
 
 
+def _iterated_value(
+    node: datastructures.ForEachData, for_port: str, body_port: str
+) -> Collection:
+    """
+    The value to scatter over one iterated port.
+
+    A for-node cannot fall back on a default here the way an atomic child can -- there
+    is nothing to iterate -- so an unfilled port is reported directly, rather than
+    leaking a bare ``NotData`` into ``itertools.product``.
+    """
+    value = node.input_ports[for_port].value
+    if isinstance(value, datastructures.NotData):
+        raise ValueError(
+            f"Iterated input '{for_port}' (body port '{body_port}') has no value to "
+            f"iterate over"
+        )
+    return cast(Collection, value)
+
+
 def _run_for(
     recipe: for_recipe.ForEachRecipe, **kwargs: Any
 ) -> datastructures.ForEachData:
@@ -244,15 +272,14 @@ def _run_for(
 
     # Build iteration axes
     nested_iters = [
-        cast(Collection, node.input_ports[body_to_for[p]].value)
-        for p in recipe.nested_ports
+        _iterated_value(node, body_to_for[p], p) for p in recipe.nested_ports
     ]
     zipped_iters = [
-        cast(Collection, node.input_ports[body_to_for[p]].value)
-        for p in recipe.zipped_ports
+        _iterated_value(node, body_to_for[p], p) for p in recipe.zipped_ports
     ]
-    # Note that we simply cast iterated input values to the form we expect, and let the
-    # user pay the price if runtime data is non-compliant.
+    # Note that beyond insisting the value arrived at all, we simply cast iterated
+    # input values to the form we expect, and let the user pay the price if runtime
+    # data is non-compliant.
 
     nested_combos = list(itertools.product(*nested_iters)) if nested_iters else [()]
     if zipped_iters:
@@ -504,3 +531,65 @@ def _populate_prospective_outputs(
                     source.port
                 ].value
                 break
+
+
+def variadic_to_inputs(recipe: base_models.NodeRecipe, /, *args, **kwargs):
+    """
+    Bind ``*args`` and ``**kwargs`` onto ``recipe.inputs``, as a helper for
+    ``NodeRecipe.__call__`` implementations and the generic recipe runner.
+
+    Every input must be filled except those in ``recipe.inputs_with_defaults``, which
+    only recipes backed by an underlying python function have any of. Nothing else can
+    supply a value after the fact, so an unfilled input is simply a missing one --
+    which is exactly the invariant
+    :func:`subgraph_validation.validate_nodes_are_fully_sourced` already holds children
+    to, so validated recipes bind their own children by construction.
+
+    Binding failures raise :class:`TypeError`, mirroring python's own behaviour for
+    bad call signatures. (Deliberately not :class:`ValueError`: recipes catch
+    exceptions by type, and a try-recipe handling ``ValueError`` must not be able to
+    swallow its caller's mistake and quietly return partial data.)
+    """
+    who = f"{type(recipe).__name__}()"
+    if len(args) > len(recipe.inputs):
+        raise TypeError(
+            f"One of your {who} calls takes {len(recipe.inputs)} inputs but "
+            f"{len(args)} positional arguments were given -- its inputs are "
+            f"{recipe.inputs}"
+        )
+    inputs = {}
+    for label, val in zip(recipe.inputs, args, strict=False):
+        inputs[label] = val
+    for label, val in kwargs.items():
+        if label in inputs:
+            raise TypeError(
+                f"One of your {who} calls got multiple values for input '{label}' -- "
+                f"as a positional arg ({inputs[label]}) and as a kwarg ({val})"
+            )
+        if label in recipe.inputs:
+            inputs[label] = val
+        else:
+            raise TypeError(
+                f"One of your {who} calls got an unexpected input '{label}' -- its "
+                f"inputs are {recipe.inputs}"
+            )
+    missing = [
+        label
+        for label in recipe.inputs
+        if label not in inputs and label not in recipe.inputs_with_defaults
+    ]
+    if missing:
+        raise TypeError(
+            f"One of your {who} calls is missing {len(missing)} required "
+            f"input: {missing}"
+        )
+    return inputs
+
+
+def data_to_return(data: datastructures.NodeData):
+    """A helper for ``NodeRecipe.__call__`` implementations"""
+    returns = tuple(p.value for p in data.output_ports.values())
+    if len(returns) == 1:
+        return returns[0]
+    else:
+        return returns

tests/integration/parsers/test_parsing_composite_workflow.py

@@ -1,12 +1,19 @@
+import dataclasses
 import inspect
 import unittest
 
 from pyiron_snippets import versions
 
-from flowrep import std
+from flowrep import std, wfms
 from flowrep.compiler import source
 from flowrep.parsers import atomic_parser, workflow_parser
-from flowrep.prospective import workflow_recipe
+from flowrep.prospective import (
+    for_recipe,
+    if_recipe,
+    try_recipe,
+    while_recipe,
+    workflow_recipe,
+)
 
 from flowrep_static import library, makers
 
@@ -264,11 +271,7 @@ def full_composite(x, /, y, *, bound):
         },
         "output_edges": {"result": "identity_0.x"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_composite_workflow",
-                "qualname": "full_composite",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(full_composite)),
             "inputs_with_defaults": [],
             "restricted_input_kinds": {
                 "x": "POSITIONAL_ONLY",
@@ -316,5 +319,111 @@ def test_roundtrip_back_to_python(self):
         )
 
 
+# =====================================================================
+# Calling the recipes instead of writing the python
+# =====================================================================
+
+# The static recipe above already spells out one recipe of every flow-control type,
+# nested inside each other. Pull them back out and call them directly.
+
+_try_flow = full_composite_node.nodes["try_0"]
+_while_flow = _try_flow.try_node.recipe.nodes["while_0"]
+_for_flow = _while_flow.case.body.recipe.nodes["for_each_0"]
+_for_body = _for_flow.body_node.recipe  # A reference-free workflow
+_if_flow = _for_body.nodes["if_0"]
+
+
+@workflow_parser.workflow
+def try_by_recipe_call(x, /, y, *, bound):
+    """:func:`full_composite` with its ``try`` block replaced by a recipe call."""
+    a = std.add(x, y)
+    b, z = _try_flow(a, y, bound)
+    result = std.identity(z)
+    return result
+
+
+@workflow_parser.workflow
+def all_flows_by_recipe_call(x, /, y, *, bound):
+    """One call to every recipe type, chained so that each feeds the next.
+
+    Not equivalent to any of the functions above -- the point is only that a single
+    workflow exercises all five ``__call__`` implementations at once.
+    """
+    a = std.add(x, y)
+    b, z = _try_flow(a, y, bound)
+    n = _while_flow(b, bound, y)
+    rs = library.my_range(y)
+    acc = _for_flow(y, n, rs)
+    s = my_sum(acc)
+    v = _if_flow(s, y, n)
+    w = _for_body(v, y, n)
+    result = std.add(z, w)
+    return result
+
+
+class TestCallingCompositeRecipes(unittest.TestCase):
+    """
+    Integration test that recipes are callable: a workflow whose nodes are invoked as
+    recipe objects must give the same answer run as plain python and run by the WfMS.
+    """
+
+    _CASES = [(1, 2, 10), (3, 1, 8)]
+
+    def _via_wfms(self, func, x, y, bound):
+        data = wfms.run_recipe(func.flowrep_recipe, x=x, y=y, bound=bound)
+        return data.output_ports["result"].value
+
+    def test_try_call_matches_python_syntax(self):
+        """Calling the try-recipe stands in for writing the ``try`` block by hand."""
+        for x, y, bound in self._CASES:
+            with self.subTest(x=x, y=y, bound=bound):
+                self.assertEqual(
+                    try_by_recipe_call(x, y, bound=bound),
+                    full_composite(x, y, bound=bound),
+                )
+
+    def test_try_call_matches_wfms(self):
+        for x, y, bound in self._CASES:
+            with self.subTest(x=x, y=y, bound=bound):
+                self.assertEqual(
+                    try_by_recipe_call(x, y, bound=bound),
+                    self._via_wfms(try_by_recipe_call, x, y, bound),
+                )
+
+    def test_all_flows_call_matches_wfms(self):
+        for x, y, bound in self._CASES:
+            with self.subTest(x=x, y=y, bound=bound):
+                self.assertEqual(
+                    all_flows_by_recipe_call(x, y, bound=bound),
+                    self._via_wfms(all_flows_by_recipe_call, x, y, bound),
+                )
+
+    def test_underfilled_call_is_not_swallowed_by_the_handler(self):
+        """``bound`` is consumed by the while-condition, three levels inside a try
+        that handles ValueError. A missing input must not be mistaken for a domain
+        error and quietly answered with the except branch."""
+        with self.assertRaises(TypeError) as ctx:
+            _try_flow(3, 2)
+        self.assertEqual(
+            str(ctx.exception),
+            "One of your TryRecipe() calls is missing 1 required input: ['bound']",
+        )
+
+    def test_every_flow_control_type_is_called(self):
+        """Guard the premise of :func:`all_flows_by_recipe_call`: if a refactor of the
+        static recipe above changes what gets pulled out, the test above could quietly
+        stop covering a recipe type."""
+        self.assertIsInstance(_try_flow, try_recipe.TryRecipe)
+        self.assertIsInstance(_while_flow, while_recipe.WhileRecipe)
+        self.assertIsInstance(_for_flow, for_recipe.ForEachRecipe)
+        self.assertIsInstance(_if_flow, if_recipe.IfRecipe)
+        self.assertIsInstance(_for_body, workflow_recipe.WorkflowRecipe)
+        self.assertIsNone(
+            _for_body.reference,
+            msg="A referenced workflow would defer to its python function instead of "
+            "running its own graph",
+        )
+
+
 if __name__ == "__main__":
     unittest.main()