Sets up is_expander and is_collect

This preps us for handling failure through our GracefulErrorAdapter
better as we can react to the node type.

hamilton/execution/executors.py

@@ -66,7 +66,7 @@ def new_callable(*args, _callable=None, **kwargs):
 
 def _modify_callable(node_source: node.NodeType, callabl: Callable):
     """This is a bit of a shortcut -- we modify the callable here as
-    we want to allow `Parallelizable[]` nodes to return a generaot
+    we want to allow `Parallelizable[]` nodes to return a generator
 
     :param node_source:
     :param callabl:

hamilton/htypes.py

@@ -283,6 +283,10 @@ class Parallelizable(typing.Generator[U, None, None], ABC):
     pass
 
 
+def is_parallelizable_type(type_: Type) -> bool:
+    return issubclass(type_, Parallelizable)
+
+
 class Collect(Generator[V, None, None], ABC):
     pass
 

hamilton/lifecycle/api.py

@@ -520,6 +520,8 @@ def do_node_execute(
             node_callable=node_.callable,
             node_kwargs=kwargs,
             task_id=task_id,
+            is_expand=node_.node_role == node.NodeType.EXPAND,
+            is_collect=node_.node_role == node.NodeType.COLLECT,
         )
 
     @abc.abstractmethod
@@ -531,6 +533,8 @@ def run_to_execute_node(
         node_callable: Any,
         node_kwargs: Dict[str, Any],
         task_id: Optional[str],
+        is_expand: bool,
+        is_collect: bool,
         **future_kwargs: Any,
     ) -> Any:
         """This method is responsible for executing the node and returning the result.
@@ -540,6 +544,8 @@ def run_to_execute_node(
         :param node_callable: Callable of the node.
         :param node_kwargs: Keyword arguments to pass to the node.
         :param task_id: The ID of the task, none if not in a task-based environment
+        :param is_expand: Whether the node is parallelizable.
+        :param is_collect: Whether the node is a collect node.
         :param future_kwargs: Additional keyword arguments -- this is kept for backwards compatibility
         :return: The result of the node execution -- up to you to return this.
         """

hamilton/lifecycle/default.py

@@ -570,8 +570,19 @@ def never_reached(wont_proceed: int) -> int:
         self.sentinel_value = sentinel_value
 
     def run_to_execute_node(
-        self, *, node_callable: Any, node_kwargs: Dict[str, Any], **future_kwargs: Any
+        self,
+        *,
+        node_callable: Any,
+        node_kwargs: Dict[str, Any],
+        is_expand: bool,
+        is_collect: bool,
+        **future_kwargs: Any,
     ) -> Any:
+        # You can use the `is_expand` to see if the node is parallelizable
+        # You can use the `is_collect` to see if the node is a collect node
+        # TODO -- if it is parallelizable, run the generator in a special way (E.G. loop through the node callable
+        # and truncate it/provide sentinels for every failure)
+        # TODO -- decide what to do with collect
         """Executes a node. If the node fails, returns the sentinel value."""
         for key, value in node_kwargs.items():
             if value == self.sentinel_value:  # == versus is

hamilton/version.py

@@ -1 +1 @@
-VERSION = (1, 69, 0)
+VERSION = (1, 70, "0rc0")