ENH(Net,TCs): Nest pipelines with NestArgs to hooks

+ doc(terms): split mest/mege terms

Author: ankostis

docs/source/arch.rst

@@ -61,28 +61,33 @@ Architecture
             - Use :func:`.compose()` factory to build :class:`.NetworkOperation`
               instances (a.k.a. pipelines).
 
-    operation merging
-    operation nesting
+    combine pipelines
         When `operation`\s and/or `pipeline`\s are `compose`\d together, there are
-        two ways to combine the operations contained into the new pipeline,
-        controlled by the ``nest`` parameter of :func:`.compose()` factory
-        :small:`(read its docstring for details on the accepted values)`:
+        two ways to combine the operations contained into the new pipeline:
+        `operation merging` (default) and `operation nesting`.
 
-        merging
-            (default) any identically-named operations override each other,
-            with the operations added earlier in the ``.compose()`` call
-            (further to the left) winning over those added later (further to the right).
+        They are selected by the ``nest`` functional parameter of :func:`.compose()`
+        factory.
+
+    operation merging
+        The default method to `combine pipelines`, also applied when simply merging `operation`\s.
 
-            :seealso: :ref:`operation-merging`
+        Any identically-named operations override each other,
+        with the operations added earlier in the ``.compose()`` call
+        (further to the left) winning over those added later (further to the right).
+
+        :seealso: :ref:`operation-merging`
+
+    operation nesting
+        The elaborate method to `combine pipelines` forming *clusters*.
 
-        nesting
-            the original pipelines are preserved intact in "isolated" clusters,
-            by prefixing the names of their operations (and optionally data)
-            by the name of the respective original pipeline that contained them
-            (or the user defines the renames).
+        The original pipelines are preserved intact in "isolated" clusters,
+        by prefixing the names of their operations (and optionally data)
+        by the name of the respective original pipeline that contained them
+        (or the user defines the renames).
 
-            :seealso: :ref:`operation-nesting`, :func:`.nest_any_node()`,
-                :func:`.dep_renamed()`, :attr:`.PlotArgs.clusters`
+        :seealso: :ref:`operation-nesting`, :func:`.compose`, :class:`.NestArgs`,
+            :func:`.nest_any_node()`, :func:`.dep_renamed()`, :attr:`.PlotArgs.clusters`
 
     compile
     compilation

docs/source/pipelines.rst

@@ -129,10 +129,9 @@ Extending pipelines
 Sometimes we begin with existing computation graph(s) to which we want to extend
 with other operations and/or pipelines.
 
-There are 2 ways to combine operations together, :term:`merging
+There are 2 ways to :term:`combine pipelines` together, :term:`merging
 <operation merging>` (the default) and :term:`nesting <operation nesting>`.
 
-
 .. _operation-merging:
 
 Merging
@@ -202,12 +201,12 @@ Let's suppose we want to break the isolation, and have all sub-pipelines
 consume & produce from a common "backlog" (n.b. in real life, we would have
 a "feeder" & "collector" operations).
 
-We do that by passing a :func:`.callable` as the ``nest`` parameter,
-which will decide which of the nodes of the original pipeline, both operations & data,
-should be prefixed (see :func:`.compose` for the exact specs of that param):
+We do that by passing as the ``nest`` parameter a :func:`.callable` which will decide
+which names of the original pipeline (operations & dependencies) should be prefixed
+(see also :func:`.compose` & :class:`.NestArgs` for how to use that param):
 
-    >>> def rename_predicate(p, node, parent):
-    ...     if node not in ("backlog", "tasks done", "todos"):
+    >>> def rename_predicate(nest_args):
+    ...     if nest_args.name not in ("backlog", "tasks done", "todos"):
     ...         return True
 
     >>> week = compose("week", *weekdays, nest=rename_predicate)

graphtik/network.py

@@ -6,7 +6,17 @@
 import logging
 from collections import abc, defaultdict
 from itertools import count
-from typing import Any, Callable, Collection, List, Mapping, Optional, Tuple, Union
+from typing import (
+    Any,
+    Callable,
+    Collection,
+    List,
+    Mapping,
+    NamedTuple,
+    Optional,
+    Tuple,
+    Union,
+)
 
 import networkx as nx
 from boltons.setutils import IndexedSet as iset
@@ -599,6 +609,20 @@ def compile(
         return plan
 
 
+class NestArgs(NamedTuple):
+    """:term:`operation nesting` callables receive instances of this class."""
+
+    #: the parent :class:`.NetworkOperation` of the operation currently being processed
+    parent: "NetworkOperation"
+    #: what is currently being renamed,
+    #: one of the string: ``(op | needs | provides)``
+    typ: str
+    #: the operation currently being processed
+    op: Operation
+    # the name of the item to be renamed/nested
+    name: str
+
+
 def build_network(
     operations,
     rescheduled=None,
@@ -608,7 +632,12 @@ def build_network(
     nest=None,
     node_props=None,
 ):
-    """The :term:`network` factory that does :term:`operation merging` before constructing it. """
+    """
+    The :term:`network` factory that does :term:`operation merging` before constructing it.
+
+    :param nest:
+        see same-named param in :func:`.compose`
+    """
     from boltons.setutils import IndexedSet as iset
     from .op import NULL_OP
     from .pipeline import NetworkOperation
@@ -622,7 +651,7 @@ def proc_op(op, parent=None):
         #
         if (
             node_props
-            or (nest and parent)
+            or nest
             or rescheduled is not None
             or endured is not None
             or parallel is not None
@@ -646,43 +675,51 @@ def proc_op(op, parent=None):
             ## If `nest`, rename the op & data (predicated by `nest`)
             #  by prefixing them with their parent netop.
             #
+            nest_args = NestArgs(parent, op, None, None)
             if nest:
-                kw["name"] = node_renamed("operation", op, parent)
-                kw["needs"] = [node_renamed("needs", n, parent) for n in op.needs]
+                kw["name"] = rename(nest_args._replace(typ="op", name=op.name))
+                kw["needs"] = [
+                    rename(nest_args._replace(typ="needs", name=n)) for n in op.needs
+                ]
                 kw["provides"] = [
-                    node_renamed("provides", n, parent) for n in op.provides
+                    rename(nest_args._replace(typ="provides", name=n))
+                    for n in op.provides
                 ]
 
             op = op.withset(**kw)
 
         return op
 
-    def node_renamed(typ, node, parent) -> str:
+    def rename(nest_args: NestArgs) -> str:
         """Handle user's or default `nest` callable's results."""
-        assert callable(nest), (nest, operations)
-
-        ret = nest(typ, node, parent)
+        assert callable(nest), (nest, nest_args)
+
+        ok = False
+        try:
+            ret = nest(nest_args)
+            if not ret:
+                # A falsy means don't touch the node.
+                ret = nest_args.name
+            elif not isinstance(ret, str):
+                # Truthy but not str values mean apply default nesting.
+                ret = nest_any_node(nest_args)
+
+            ok = True
+            return ret
+        finally:
+            if not ok:
+                log.warning("Failed to nest-rename %s", nest_args)
 
-        if not ret:
-            # A falsy means don't touch the node.
-            return node.name if isinstance(node, Operation) else node
-
-        if not isinstance(ret, str):
-            # Truthy but not str values mean nest!
-            ret = nest_any_node(typ, node, parent)
-
-        return ret
-
-    ## Set default nesting if requested by user.
-    #
     if nest:
+        ## Set default nesting if not one provided by user.
+        #
         if not callable(nest):
             nest = nest_any_node
 
     merge_set = iset()  # Preseve given node order.
     for op in operations:
         if isinstance(op, NetworkOperation):
-            merge_set.update(proc_op(s, op.name) for s in yield_ops(op.net.graph))
+            merge_set.update(proc_op(s, op) for s in yield_ops(op.net.graph))
         else:
             merge_set.add(proc_op(op))
     merge_set = iset(i for i in merge_set if not isinstance(i, NULL_OP))
@@ -693,22 +730,18 @@ def node_renamed(typ, node, parent) -> str:
     return net
 
 
-def nest_any_node(
-    node_type: str, node: Union[Operation, str], parent: "NetworkOperation"
-) -> str:
-    """Nest both operation & data `node` under `parent` (if given).
+def nest_any_node(nest_args: NestArgs) -> str:
+    """Nest both operation & data under `parent`'s name (if given).
 
     :return:
         the nested name of the operation or data
     """
 
     def prefixed(name):
-        return f"{parent}.{name}" if parent else name
+        return f"{nest_args.parent.name}.{name}" if nest_args.parent else name
 
-    if isinstance(node, Operation):
-        new_name = prefixed(node.name)
-    else:
-        assert isinstance(node, str), locals()
-        new_name = dep_renamed(node, prefixed)
-
-    return new_name
+    return (
+        prefixed(nest_args.name)
+        if nest_args.typ == "op"
+        else dep_renamed(nest_args.name, prefixed)
+    )

graphtik/pipeline.py

@@ -331,7 +331,7 @@ def compose(
     endured=None,
     parallel=None,
     marshalled=None,
-    nest: Union[Callable[[str], str], Union[bool, str]] = None,
+    nest: Union[Callable[["NestArgs"], str], Union[bool, str]] = None,
     node_props=None,
 ) -> NetworkOperation:
     """
@@ -350,29 +350,26 @@ def compose(
         ``operation``.
     :param nest:
         - If false (default), applies :term:`operation merging`, not *nesting*.
-        - if true, applies :term:`operation nesting` to :class:`.FunctionalOperation`\\s
-          only (performed by :func:`.op.nest_any_node()`;
-        - if it is :func:`.callable`, it is given each node and the parent pipeline
-          (if combining pipeline(s)) to decide the node's name.
+        - if true, applies :term:`operation nesting` to :all types of nodes
+          (performed by :func:`.nest_any_node()`;
+        - if it is a :func:`.callable`, it is given a :class:`.NestArgs` instance
+          to decide the node's name.
 
-          The callable may return the new-name (str), or a true/false to apply
-          the default renaming which is to rename all nodes (as performed by
-          :func:`.op.nest_any_node()`).
+          The callable may return a *str* for the new-name, or any other true/false
+          to apply default nesting which is to rename all nodes, as performed
+          by :func:`.nest_any_node()`.
 
-          For example, to nest just the operations, call::
+          For example, to nest just operation's names (but not their dependencies),
+          call::
 
               compose(
                   ...,
-                  nest=lambda typ, node, parent: isinstance(n, Operation)
+                  nest=lambda nest_args: nest_args.typ == "op"
               )
 
-          ...where `typ` is one of ``(operation | needs | provides)`` strings.
-
-          Of course the callable can apply a renaming irrelevant from the `parent`.
-
           .. Attention::
-              The callable must preserve any :term:`modifier` on data nodes,
-              so :func:`.dep_renamed` should be used.
+              The callable SHOULD wish to preserve any :term:`modifier` on dependencies,
+              and use :func:`.dep_renamed`.
 
           :seealso: :ref:`operation-nesting` for examples
 

test/test_graphtik.py

@@ -280,9 +280,25 @@ def powers_in_range(a, exponent):
     assert sol == exp
 
 
+def test_network_nest_errors(caplog):
+    def screamy_nester(nest_args):
+        raise RuntimeError("Bluff")
+
+    with pytest.raises(RuntimeError, match="Bluff"):
+        compose(
+            "test_nest_err",
+            operation(str, "op1"),
+            operation(str, "op2"),
+            nest=screamy_nester,
+        )
+    for record in caplog.records:
+        if record.levelname == "WARNING":
+            assert "(parent=None, typ='op', op=None, name='op1')" in record.message
+
+
 def test_network_nest_ops_only():
-    def ops_only(t, n, p):
-        return isinstance(n, Operation)
+    def ops_only(nest_args):
+        return nest_args.typ == "op"
 
     sum_op1 = operation(name="sum_op1", needs=["a", "b"], provides="sum1")(add)
     sum_op2 = operation(name="sum_op2", needs=["a", "b"], provides="sum2")(add)
@@ -397,11 +413,11 @@ def test_network_merge_in_doctests():
     week = compose("week", *weekdays, nest=True)
     assert len(week.ops) == 6
 
-    def rename_predicate(typ, node, parent):
-        if node not in ("backlog", "tasks done", "todos"):
+    def nester(nest_args):
+        if nest_args.name not in ("backlog", "tasks done", "todos"):
             return True
 
-    week = compose("week", *weekdays, nest=rename_predicate)
+    week = compose("week", *weekdays, nest=nester)
     assert len(week.ops) == 6
     sol = week.compute({"backlog": "a lot!"})
     assert sol == {
@@ -1623,7 +1639,7 @@ def test_combine_networks(exemethod, bools):
             name="sub2", needs=["a_minus_ab", "c"], provides="a_minus_ab_minus_c"
         )(sub),
         parallel=parallel2,
-        nest=lambda t, n, p: isinstance(n, Operation),
+        nest=lambda nest_args: nest_args.typ == "op",
     )
     ## Ensure all old-nodes were prefixed.
     #