Adds method based options, limited to one, to allow nodes based on a specific method (think filter, join, etc)...

Reimplementation of "Filter", with (rather simple) code from rdc.etl.

bonobo/config/__init__.py

@@ -1,5 +1,5 @@
 from bonobo.config.configurables import Configurable
-from bonobo.config.options import Option
+from bonobo.config.options import Option, Method
 from bonobo.config.processors import ContextProcessor
 from bonobo.config.services import Container, Service
 
@@ -8,5 +8,6 @@ __all__ = [
     'Container',
     'ContextProcessor',
     'Option',
+    'Method',
     'Service',
 ]

bonobo/config/configurables.py

@@ -1,5 +1,6 @@
+from bonobo.config.options import Method, Option
 from bonobo.config.processors import ContextProcessor
-from bonobo.config.options import Option
+from bonobo.errors import ConfigurationError
 
 __all__ = [
     'Configurable',
@@ -17,6 +18,7 @@ class ConfigurableMeta(type):
         cls.__options__ = {}
         cls.__positional_options__ = []
         cls.__processors__ = []
+        cls.__wrappable__ = None
 
         for typ in cls.__mro__:
             for name, value in typ.__dict__.items():
@@ -24,6 +26,10 @@ class ConfigurableMeta(type):
                     if isinstance(value, ContextProcessor):
                         cls.__processors__.append(value)
                     else:
+                        if isinstance(value, Method):
+                            if cls.__wrappable__:
+                                raise ConfigurationError('Cannot define more than one "Method" option in a configurable. That may change in the future.')
+                            cls.__wrappable__ = name
                         if not value.name:
                             value.name = name
                         if not name in cls.__options__:
@@ -43,6 +49,13 @@ class Configurable(metaclass=ConfigurableMeta):
 
     """
 
+    def __new__(cls, *args, **kwargs):
+        if cls.__wrappable__ and len(args) == 1 and hasattr(args[0], '__call__'):
+            wrapped, args = args[0], args[1:]
+            return type(wrapped.__name__, (cls, ), {cls.__wrappable__: wrapped})
+
+        return super().__new__(cls)
+
     def __init__(self, *args, **kwargs):
         super().__init__()
 
@@ -90,3 +103,6 @@ class Configurable(metaclass=ConfigurableMeta):
         """ You can implement a configurable callable behaviour by implemenenting the call(...) method. Of course, it is also backward compatible with legacy __call__ override.
         """
         return self.call(*args, **kwargs)
+
+    def call(self, *args, **kwargs):
+        raise NotImplementedError('Not implemented.')

bonobo/config/options.py

@@ -16,13 +16,34 @@ class Option:
         self._creation_counter = Option._creation_counter
         Option._creation_counter += 1
 
-    def get_default(self):
-        return self.default() if callable(self.default) else self.default
-
     def __get__(self, inst, typ):
         if not self.name in inst.__options_values__:
             inst.__options_values__[self.name] = self.get_default()
         return inst.__options_values__[self.name]
 
+    def __set__(self, inst, value):
+        inst.__options_values__[self.name] = self.clean(value)
+
+    def get_default(self):
+        return self.default() if callable(self.default) else self.default
+
+    def clean(self, value):
+        return self.type(value) if self.type else value
+
+
+class Method(Option):
+    def __init__(self):
+        super().__init__(None, required=False, positional=True)
+
+    def __get__(self, inst, typ):
+        if not self.name in inst.__options_values__:
+            inst.__options_values__[self.name] = getattr(inst, self.name)
+        return inst.__options_values__[self.name]
+
     def __set__(self, inst, value):
         inst.__options_values__[self.name] = self.type(value) if self.type else value
+
+    def clean(self, value):
+        if not hasattr(value, '__call__'):
+            raise ValueError('{} value must be callable.'.format(type(self).__name__))
+        return value

bonobo/errors.py

@@ -52,3 +52,7 @@ class ValidationError(RuntimeError):
 
 class ProhibitedOperationError(RuntimeError):
     pass
+
+
+class ConfigurationError(Exception):
+    pass

bonobo/examples/utils/filter.py

@@ -0,0 +1,21 @@
+import bonobo
+
+from bonobo.filter import Filter
+
+
+class OddOnlyFilter(Filter):
+    def filter(self, i):
+        return i % 2
+
+
+@Filter
+def MultiplesOfThreeOnlyFilter(self, i):
+    return not (i % 3)
+
+
+graph = bonobo.Graph(
+    lambda: tuple(range(50)),
+    OddOnlyFilter(),
+    MultiplesOfThreeOnlyFilter(),
+    print,
+)

bonobo/execution/graph.py

@@ -26,11 +26,7 @@ class GraphExecutionContext:
         self.services = Container(services) if services else Container()
 
         for i, node_context in enumerate(self):
-            try:
-                node_context.outputs = [self[j].input for j in self.graph.outputs_of(i)]
-            except KeyError:
-                continue
-
+            node_context.outputs = [self[j].input for j in self.graph.outputs_of(i)]
             node_context.input.on_begin = partial(node_context.send, BEGIN, _control=True)
             node_context.input.on_end = partial(node_context.send, END, _control=True)
             node_context.input.on_finalize = partial(node_context.stop)

bonobo/filter/__init__.py

@@ -0,0 +1,28 @@
+from bonobo.constants import NOT_MODIFIED
+
+from bonobo.config import Configurable, Method
+
+
+class Filter(Configurable):
+    """Filter out hashes from the stream depending on the :attr:`filter` callable return value, when called with the
+    current hash as parameter.
+
+    Can be used as a decorator on a filter callable.
+
+    .. attribute:: filter
+
+        A callable used to filter lines.
+        
+        If the callable returns a true-ish value, the input will be passed unmodified to the next items.
+        
+        Otherwise, it'll be burnt.
+
+    """
+
+    filter = Method()
+
+    def call(self, *args, **kwargs):
+        if self.filter(*args, **kwargs):
+            return NOT_MODIFIED
+
+

tests/test_config_method.py

@@ -0,0 +1,73 @@
+import pytest
+
+from bonobo.config import Configurable, Method, Option
+from bonobo.errors import ConfigurationError
+
+
+class MethodBasedConfigurable(Configurable):
+    handler = Method()
+    foo = Option(positional=True)
+    bar = Option()
+
+    def call(self, *args, **kwargs):
+        self.handler(*args, **kwargs)
+
+
+def test_one_wrapper_only():
+    with pytest.raises(ConfigurationError):
+        class TwoMethods(Configurable):
+            h1 = Method()
+            h2 = Method()
+
+
+def test_define_with_decorator():
+    calls = []
+
+    @MethodBasedConfigurable
+    def Concrete(self, *args, **kwargs):
+        calls.append((args, kwargs,))
+
+    t = Concrete('foo', bar='baz')
+    assert len(calls) == 0
+    t()
+    assert len(calls) == 1
+
+
+def test_define_with_argument():
+    calls = []
+
+    def concrete_handler(*args, **kwargs):
+        calls.append((args, kwargs,))
+
+    t = MethodBasedConfigurable('foo', bar='baz', handler=concrete_handler)
+    assert len(calls) == 0
+    t()
+    assert len(calls) == 1
+
+def test_define_with_inheritance():
+    calls = []
+
+    class Inheriting(MethodBasedConfigurable):
+        def handler(self, *args, **kwargs):
+            calls.append((args, kwargs,))
+
+    t = Inheriting('foo', bar='baz')
+    assert len(calls) == 0
+    t()
+    assert len(calls) == 1
+
+
+def test_inheritance_then_decorate():
+    calls = []
+
+    class Inheriting(MethodBasedConfigurable):
+        pass
+
+    @Inheriting
+    def Concrete(self, *args, **kwargs):
+        calls.append((args, kwargs,))
+
+    t = Concrete('foo', bar='baz')
+    assert len(calls) == 0
+    t()
+    assert len(calls) == 1

tests/test_execution.py

@@ -59,11 +59,24 @@ def test_simple_execution_context():
         assert ctx[i].wrapped is node
 
     assert not ctx.alive
+    assert not ctx.started
+    assert not ctx.stopped
 
     ctx.recv(BEGIN, Bag(), END)
 
     assert not ctx.alive
+    assert not ctx.started
+    assert not ctx.stopped
 
     ctx.start()
 
     assert ctx.alive
+    assert ctx.started
+    assert not ctx.stopped
+
+    ctx.stop()
+
+    assert not ctx.alive
+    assert ctx.started
+    assert ctx.stopped
+