I'm trying to build a framework similar to Scrapy ItemPipelines or Pipeless. The motivation is to be able to build generic data pipelines via defining a modular collection of "pipe" classes that handle distinct steps within the pipeline. I've taken some tips from here. I'm looking for general feedback.


from zope.interface import Interface, Attribute

class IPipe(Interface):
    """A pipe is responsible for processing a single item
    and yields one or more iterables
    def process_item(item):
        """process single item

class IPipeline(Interface):
    """ Master pipeline interface

    pipe_runners = Attribute('a collection of pipe_runners')

    def run(it):
        """begin pipeline execution given iterable

class IPipelineRunner(Interface):
    """Handles execution of pipe and skip item exceptions

    pipe = Attribute('single pipe being wrapped')

    def execute(it):
        """execute pipe with given iterable


from pipeable.interfaces import IPipeline, IPipe, IPipelineRunner
from pipeable.exceptions import InvalidPipeInput, SkipPipeItem
from zope.interface import implements, implementer
import collections

class Pipeline(object):

    def __init__(self, pipes):
        #xxx TODO handle list of pipes of now, later handle dict with order numbers, and yaml file constructor params
        self.pipe_runners = []
        for pipe in pipes:
            pipe_runner = PipeRunner(implementer(IPipe)(pipe()))

    def run(self, it):
        if not hasattr(it, '__iter__'):
            raise InvalidPipeInput("Pipeline requires iterable input, got: " + it)
        res = it
        for pipe_runner in self.pipe_runners:
            res = pipe_runner.execute(res)
        return res

class PipeRunner(object):

    def __init__(self, pipe):
        self.pipe = pipe

    def execute(self, item_gen):            
        for item in item_gen:
                for res in self.pipe.process_item(item):
                    #process_items can produce multiple yields
                    yield res
            except SkipPipeItem:


import unittest

class TestPipeline(unittest.TestCase):

    def _getTargetClass(self):
        from pipeable.components import Pipeline
        return Pipeline

    def _makeOne(self, pipes):
        return self._getTargetClass()(pipes)

    def test_class_conforms_to_IPipeline(self):
        from zope.interface.verify import verifyClass
        from pipeable.interfaces import IPipeline
        verifyClass(IPipeline, self._getTargetClass())

    def test_should_run_simple_pipes_add_one_add_two(self):
        test_pipes = self._getSimpleTestPipes()
        pipe_line = self._makeOne(test_pipes)
        res_gen = pipe_line.run([1])
        res = res_gen.next()
        self.assertEquals(4, res, res)

    def test_should_yield_forward_and_backward(self):
        test_pipes = self._getForwardReverseTitles()
        pipe_line = self._makeOne(test_pipes)
        res = pipe_line.run(['foo', 'bar'])
        self.assertEquals(['Foo', 'Oof', 'Bar', 'Rab'], 
                          list(res), list(res))

    def test_should_skip_foo(self):
        from pipeable.exceptions import SkipPipeItem
        class ForwardReverseNoFoo(object):
            def process_item(self, str_item):
                if str_item == 'foo':
                    raise SkipPipeItem
                yield str_item
                yield str_item[::-1]
        test_pipes = [ForwardReverseNoFoo]
        pipe_line = self._makeOne(test_pipes)
        res = pipe_line.run(['foo', 'bar'])
        self.assertEquals(['bar', 'rab'], 
                          list(res), list(res))

    def test_should_raise_invalid_pipe_input_on_non_iterable(self):
        from pipeable.exceptions import InvalidPipeInput
        test_pipes = self._getForwardReverseTitles()
        pipe_line = self._makeOne(test_pipes)
        self.assertRaises(InvalidPipeInput, pipe_line.run,

    def _getSimpleTestPipes(self):
        class TestPipe1(object):
            def process_item(self, item):
                yield item + 1
        class TestPipe2(object):
            def process_item(self, item):
                yield item + 2
        return [TestPipe1, TestPipe2]

    def _getForwardReverseTitles(self):
        class ForwardReverse(object):
            def process_item(self, str_item):
                yield str_item
                yield str_item[::-1]
        class Title(object):
            def process_item(self, str_item):
                yield str_item.title()
        return [ForwardReverse, Title]
  • 1
    \$\begingroup\$ Your Pipeline.run() requires an iterable, but you then proceed to hand that iterable to multiple pipeline runners; this won't work because an iterable can only be iterated over once. \$\endgroup\$ – Martijn Pieters Nov 13 '13 at 16:02
  • 1
    \$\begingroup\$ I am not certain why you support multiple pipelines in the IPipeline interface here either, and why the runner needs to be separated. Won't a simpler IPipeline interface that can be run directly, containing just one actual sequence of pipes, be easier to understand and use? \$\endgroup\$ – Martijn Pieters Nov 13 '13 at 16:03
  • 1
    \$\begingroup\$ Also, in my original post I talked about the need to have a central 'blackboard', to maintain shared state between pipe segments for a given run of the pipeline. Your IPipe interface has no access to anything else; should implementations take care of referencing a per-pipeline global state themselves? Without a pipeline context, they have no means to distinguish one run from another. \$\endgroup\$ – Martijn Pieters Nov 13 '13 at 16:07
  • \$\begingroup\$ Thanks for the feedback @MartijnPieters . In the latest iteration I believe I've simplified it, as well as added access to a pipeline context. github.com/bcajes/pipeable \$\endgroup\$ – Flack Dec 7 '13 at 16:55

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.