Making config/util/structs apis available at level 2 import (x.y), implements the roots for loopbacks (recursive transformations). This still needs work, as its hard not to close an input queue as soon as the last item was read.

bonobo/_api.py

@@ -1,6 +1,6 @@
 import logging
 
-from bonobo.structs import Bag, Graph, Token
+from bonobo.structs import Bag, ErrorBag, Graph, Token
 from bonobo.nodes import CsvReader, CsvWriter, FileReader, FileWriter, Filter, JsonReader, JsonWriter, Limit, \
     PickleReader, PickleWriter, PrettyPrinter, RateLimited, Tee, arg0_to_kwargs, count, identity, kwargs_to_arg0, noop
 from bonobo.strategies import create_strategy
@@ -70,7 +70,7 @@ def run(graph, strategy=None, plugins=None, services=None):
 
 
 # bonobo.structs
-register_api_group(Bag, Graph, Token)
+register_api_group(Bag, ErrorBag, Graph, Token)
 
 # bonobo.strategies
 register_api(create_strategy)

bonobo/constants.py

@@ -3,6 +3,7 @@ from bonobo.structs.tokens import Token
 BEGIN = Token('Begin')
 END = Token('End')
 INHERIT_INPUT = Token('InheritInput')
+LOOPBACK = Token('Loopback')
 NOT_MODIFIED = Token('NotModified')
 DEFAULT_SERVICES_FILENAME = '_services.py'
 DEFAULT_SERVICES_ATTR = 'get_services'

bonobo/execution/node.py

@@ -2,13 +2,13 @@ import traceback
 from queue import Empty
 from time import sleep
 
-from bonobo.constants import INHERIT_INPUT, NOT_MODIFIED
+from bonobo.constants import INHERIT_INPUT, NOT_MODIFIED, BEGIN, END
 from bonobo.errors import InactiveReadableError, UnrecoverableError
 from bonobo.execution.base import LoopingExecutionContext
 from bonobo.structs.bags import Bag
 from bonobo.structs.inputs import Input
 from bonobo.util.compat import deprecated_alias
-from bonobo.util.errors import is_error
+from bonobo.util.inspect import iserrorbag, isloopbackbag
 from bonobo.util.iterators import iter_if_not_sequence
 from bonobo.util.objects import get_name
 from bonobo.util.statistics import WithStatistics
@@ -65,8 +65,10 @@ class NodeExecutionContext(WithStatistics, LoopingExecutionContext):
         if not _control:
             self.increment('out')
 
-        if is_error(value):
+        if iserrorbag(value):
             value.apply(self.handle_error)
+        elif isloopbackbag(value):
+            self.input.put(value)
         else:
             for output in self.outputs:
                 output.put(value)
@@ -137,7 +139,7 @@ def _resolve(input_bag, output):
     if output is NOT_MODIFIED:
         return input_bag
 
-    if is_error(output):
+    if iserrorbag(output):
         return output
 
     # If it does not look like a bag, let's create one for easier manipulation

bonobo/structs/__init__.py

@@ -1,5 +1,11 @@
-from bonobo.structs.bags import Bag
+from bonobo.structs.bags import Bag, ErrorBag, LoopbackBag
 from bonobo.structs.graphs import Graph
 from bonobo.structs.tokens import Token
 
-__all__ = ['Bag', 'Graph', 'Token']
+__all__ = [
+    'Bag',
+    'ErrorBag',
+    'Graph',
+    'LoopbackBag',
+    'Token',
+]

bonobo/structs/bags.py

@@ -1,6 +1,6 @@
 import itertools
 
-from bonobo.constants import INHERIT_INPUT
+from bonobo.constants import INHERIT_INPUT, LOOPBACK
 
 __all__ = [
     'Bag',
@@ -33,8 +33,10 @@ class Bag:
     
     """
 
+    default_flags = ()
+
     def __init__(self, *args, _flags=None, _parent=None, **kwargs):
-        self._flags = _flags or ()
+        self._flags = type(self).default_flags + (_flags or ())
         self._parent = _parent
         self._args = args
         self._kwargs = kwargs
@@ -106,5 +108,9 @@ class Bag:
         )
 
 
+class LoopbackBag(Bag):
+    default_flags = (LOOPBACK,)
+
+
 class ErrorBag(Bag):
     pass

bonobo/util/__init__.py

@@ -1,6 +1,28 @@
+from bonobo.util.inspect import (
+    inspect_node,
+    isbag,
+    isconfigurable,
+    isconfigurabletype,
+    iscontextprocessor,
+    iserrorbag,
+    isloopbackbag,
+    ismethod,
+    isoption,
+    istype,
+)
 from bonobo.util.python import require
 
 # Bonobo's util API
 __all__ = [
-    'require'
+    'require',
+    'inspect_node',
+    'isbag',
+    'isconfigurable',
+    'isconfigurabletype',
+    'iscontextprocessor',
+    'iserrorbag',
+    'isloopbackbag',
+    'ismethod',
+    'isoption',
+    'istype',
 ]

bonobo/util/errors.py

@@ -1,13 +1,6 @@
 import sys
 from textwrap import indent
 
-from bonobo import settings
-from bonobo.structs.bags import ErrorBag
-
-
-def is_error(bag):
-    return isinstance(bag, ErrorBag)
-
 
 def _get_error_message(exc):
     if hasattr(exc, '__str__'):

bonobo/util/inspect.py

@@ -1,5 +1,18 @@
 from collections import namedtuple
 
+from bonobo.constants import LOOPBACK
+
+
+def isconfigurable(mixed):
+    """
+    Check if the given argument is an instance of :class:`bonobo.config.Configurable`.
+
+    :param mixed:
+    :return: bool
+    """
+    from bonobo.config.configurables import Configurable
+    return isinstance(mixed, Configurable)
+
 
 def isconfigurabletype(mixed):
     """
@@ -13,17 +26,6 @@ def isconfigurabletype(mixed):
     return isinstance(mixed, ConfigurableMeta)
 
 
-def isconfigurable(mixed):
-    """
-    Check if the given argument is an instance of :class:`bonobo.config.Configurable`.
-
-    :param mixed:
-    :return: bool
-    """
-    from bonobo.config.configurables import Configurable
-    return isinstance(mixed, Configurable)
-
-
 def isoption(mixed):
     """
     Check if the given argument is an instance of :class:`bonobo.config.Option`.
@@ -68,6 +70,38 @@ def istype(mixed):
     return isinstance(mixed, type)
 
 
+def isbag(mixed):
+    """
+    Check if the given argument is an instance of a :class:`bonobo.Bag`.
+
+    :param mixed:
+    :return: bool
+    """
+    from bonobo.structs.bags import Bag
+    return isinstance(mixed, Bag)
+
+
+def iserrorbag(mixed):
+    """
+    Check if the given argument is an instance of an :class:`bonobo.ErrorBag`.
+
+    :param mixed:
+    :return: bool
+    """
+    from bonobo.structs.bags import ErrorBag
+    return isinstance(mixed, ErrorBag)
+
+
+def isloopbackbag(mixed):
+    """
+    Check if the given argument is an instance of a :class:`bonobo.Bag`, marked for loopback behaviour.
+
+    :param mixed:
+    :return: bool
+    """
+    return isbag(mixed) and LOOPBACK in mixed.flags
+
+
 ConfigurableInspection = namedtuple(
     'ConfigurableInspection', [
         'type',