feat: support optional flag in from_ extractors

* Add optional flag to ExtractorFactory to skip data loader when target
  was not found rather than raise exceptions.
* Add unit tests for optional and non-optional extractors.
* Add more detailed comments and documentation note.

Author: mgdemers

docs/concepts/materialization.rst

@@ -92,6 +92,9 @@ Passing ``from_`` and ``to`` Apache Hamilton objects to ``Builder().with_materia
 2. Observability ✅: Loaders and savers are part of the dataflow. You can view them with ``Driver.display_all_functions()`` and execute nodes by requesting them with ``Driver.execute()``.
 3. Flexibility ✅: The loading and saving behavior is decoupled from the dataflow and can modified easily when creating the ``Driver`` and executing code.
 
+.. note::
+
+    ``from_`` data loaders can be specified as optional with ``optional=True``. This allows specified data loaders to be skipped rather than raise an exception when not referenced by a dataflow.
 
 Dynamic materializers
 ~~~~~~~~~~~~~~~~~~~~~

hamilton/io/materialization.py

@@ -136,6 +136,7 @@ def __init__(
         self,
         target: str,
         loaders: list[type[DataLoader]],
+        optional: bool = False,
         **data_loader_kwargs: Any | SingleDependency,
     ):
         """Instantiates an ExtractorFactory. Note this is not a public API -- this is
@@ -144,10 +145,14 @@ def __init__(
 
         :param target: Parameter, into which we're loading the data
         :param loaders: A list of data loaders that are viable candidates, given the key after `from_`
+        :param optional: Whether missing targets should raise exceptions or skip being loaded.
+            Optional=True will mean that if the target does not exist in the graph, we skip injecting
+            this into the graph.
         :param data_loader_kwargs: Keyword arguments for the data loaders.
         """
         self.target = target
         self.loaders = loaders
+        self.optional = optional
         self.data_loader_kwargs = process_kwargs(data_loader_kwargs)
 
     def generate_nodes(self, fn_graph: graph.FunctionGraph) -> list[node.Node]:
@@ -163,6 +168,8 @@ def generate_nodes(self, fn_graph: graph.FunctionGraph) -> list[node.Node]:
         # TODO -- add some nodes to the graph
         node_with_target = fn_graph.nodes.get(self.target)
         if node_with_target is None:
+            if self.optional:
+                return []
             raise ValueError(
                 f"Could not find node with name: {self.target} in function "
                 f"graph. Available nodes: {list(fn_graph.nodes.keys()) + [...] if len(fn_graph.nodes) > 10 else []}"

tests/io/test_materialization.py

@@ -156,6 +156,45 @@ def test(input_data: dict) -> dict:
     assert input_data_node.type == dict  # From above
 
 
+def test_extractor_factory_exception_on_missing_targets():
+    factory = ExtractorFactory(
+        "input_data",
+        loaders=[MockDataLoader],
+        optional=False,
+    )
+
+    def test() -> dict:
+        return {"loaded_value": {}}
+
+    base_node = node.Node.from_fn(test)
+    nodes_without_dependencies = graph.update_dependencies(
+        {base_node.name: base_node}, lifecycle_base.LifecycleAdapterSet(base.DefaultAdapter())
+    )
+    fn_graph = graph.FunctionGraph(nodes_without_dependencies, {})
+    with pytest.raises(ValueError):
+        factory.generate_nodes(fn_graph)
+
+
+def test_extractor_factory_optional_flag_skips_missing_targets():
+    factory = ExtractorFactory(
+        "input_data",
+        loaders=[MockDataLoader],
+        optional=True,
+    )
+
+    def test() -> dict:
+        return {"loaded_value": {}}
+
+    base_node = node.Node.from_fn(test)
+    nodes_without_dependencies = graph.update_dependencies(
+        {base_node.name: base_node}, lifecycle_base.LifecycleAdapterSet(base.DefaultAdapter())
+    )
+    fn_graph = graph.FunctionGraph(nodes_without_dependencies, {})
+    nodes = factory.generate_nodes(fn_graph)
+    nodes_by_name = {node_.name: node_ for node_ in nodes}
+    assert "input_data" not in nodes_by_name
+
+
 def test_materializer_factory_generates_nodes_with_builder():
     factory = MaterializerFactory(
         "test_materializer",