def wait_for_http_server(host, port, wait_time_seconds=0.1, max_tries=100):
    for _ in range(max_tries):
        try:
            resp = requests.get(f'http://{host}:{port}')
            return resp
        except requests.exceptions.ConnectionError:
            time.sleep(wait_time_seconds)
            continue
    raise RuntimeError(f'Failed to start HTTP server at {host}:{port}') class Var(object):
    def __init__(self, name):
        super(Var, self).__init__()
        self.name = name class NotEnoughData(Exception):
    """
    Exception thrown by extractors when they do not yet have enough data to
    execute the processing step
    """
    pass class Aggregator(object):
    """
    A mixin for Node-derived classes that addresses the case when the processing
    node cannot do its computation until all input has been received
    """

    def __init__(self, needs=None):
        super(Aggregator, self).__init__(needs=needs)

    def _dequeue(self):
        if not self._finalized:
            raise NotEnoughData()
        return super(Aggregator, self)._dequeue() class Node(object):
    def __init__(self, needs=None):
        super(Node, self).__init__()
        if not inspect.isgeneratorfunction(self._process):
            raise InvalidProcessMethod(self.__class__)

        self._cache = None
        self._listeners = []

        self._needs = dictify(needs)
        self._dependecy_names = dict(
            (id(v), k) for k, v in list(self._needs.items()))

        for n in self.dependencies:
            n.add_listener(self)

        self._finalized_dependencies = set()
        self._enqueued_dependencies = set()

    @property
    def dependencies(self):
        if isinstance(self._needs, dict):
            return list(self._needs.values())
        return self._needs

    def _dependency_name(self, pusher):
        return self._dependecy_names[id(pusher)]

    def __repr__(self):
        return self.__class__.__name__

    def __str__(self):
        return self.__repr__()

    def __enter__(self):
        return self

    def __exit__(self, t, value, traceback):
        pass

    @property
    def version(self):
        return self.__class__.__name__

    @property
    def needs(self):
        return self._needs

    @property
    def dependency_count(self):
        return len(self._needs)

    @property
    def is_root(self):
        return not self._needs

    @property
    def is_leaf(self):
        return not self._listeners

    def add_listener(self, listener):
        self._listeners.append(listener)

    def find_listener(self, predicate):
        for l in self._listeners:
            return l if predicate(l) else l.find_listener(predicate)
        return None

    def disconnect(self):
        for e in self.dependencies:
            try:
                e._listeners.remove(self)
            except ValueError:
                pass

    def _enqueue(self, data, pusher):
        self._cache = data

    def _dequeue(self):
        if self._cache is None:
            raise NotEnoughData()

        v, self._cache = self._cache, None
        return v

    def _process(self, data):
        yield data

    def _first_chunk(self, data):
        return data

    def _last_chunk(self):
        return iter(())

    def _finalize(self, pusher):
        pass

    @property
    def _finalized(self):
        """
        Return true if all dependencies have informed this node that they'll
        be sending no more data (by calling _finalize()), and that they have
        sent at least one batch of data (by calling enqueue())
        """
        return \
            len(self._finalized_dependencies) >= self.dependency_count \
            and len(self._enqueued_dependencies) >= self.dependency_count

    def _push(self, data, queue=None):
        queue.appendleft((
            id(self),
            self.process.__name__,
            {'pusher': self, 'data': data, 'queue': queue}))

    def _finish(self, pusher=None, queue=None):
        self._finalize(pusher)

        if pusher in self.dependencies:
            self._finalized_dependencies.add(id(pusher))

        if pusher:
            return
        queue.appendleft((
            id(self),
            self._finish.__name__,
            {'pusher': self, 'queue': queue}))

    def process(self, data=None, pusher=None, queue=None):
        if data is not None:
            self._enqueued_dependencies.add(id(pusher))
            self._enqueue(data, pusher)

        try:
            inp = self._dequeue()
            inp = self._first_chunk(inp)
            self._first_chunk = lambda x: x
            for d in self._process(inp):
                yield self._push(d, queue=queue)
        except NotEnoughData:
            yield None

        if self.is_root or self._finalized:
            for chunk in self._last_chunk():
                yield self._push(chunk, queue=queue)
            self._finish(pusher=None, queue=queue)
            self._push(None, queue=queue)
            yield None class InvalidProcessMethod(Exception):
    """
    Exception thrown when the _process method of an Node is not a generator
    """

    def __init__(self, cls):
        msg = '{name}._process method must be a generator' \
            .format(name=cls.__name__)
        super(InvalidProcessMethod, self).__init__(msg) class IteratorNode(Node):
    def __init__(self, needs=None):
        super(IteratorNode, self).__init__(needs=needs)

    def _process(self, data):
        for x in data:
            yield x class BaseModel(object, metaclass=MetaModel):
    def __init__(self, _id=None):
        super(BaseModel, self).__init__()
        if _id:
            self._id = _id

    def __getattribute__(self, key):
        f = object.__getattribute__(self, key)

        if not isinstance(f, Feature):
            return f

        BaseModel._ensure_persistence_settings(self.__class__)
        feature = getattr(self.__class__, key)
        decoded = feature.__call__(self._id, persistence=self.__class__)
        setattr(self, key, decoded)
        return decoded

    @classmethod
    def _build_extractor(cls, _id, **kwargs):
        g = Graph()
        for feature in list(cls.features.values()):
            feature._build_extractor(_id, g, cls, **kwargs)
        return g

    @classmethod
    def random(cls):
        all_ids = list(cls.database.iter_ids())
        return cls(_id=choice(all_ids))

    @classmethod
    def exists(cls, _id=None, feature=None):
        if feature and not feature.store:
            raise ValueError('feature must have store=True')

        if _id is None:
            try:
                _id = cls.id_provider.new_id()
            except:
                raise ValueError(
                    '_id must be provided explicitly, or it must be static')

        feature = feature or [f for f in cls.iter_features() if f.store][0]
        feature_key = feature.feature_key(_id, cls)
        return feature_key in cls.database

    @classmethod
    def process(cls, raise_if_exists=False, **kwargs):
        BaseModel._ensure_persistence_settings(cls)
        _id = cls.id_provider.new_id(**kwargs)

        if raise_if_exists and cls.exists(_id):
            # # choose a random, stored feature
            # feature = filter(lambda f: f.store, cls.iter_features())[0]
            # feature_key = feature.feature_key(_id, cls)
            # # check if that feature is already stored
            # if feature_key in cls.database:
            raise ModelExistsError(
                '{_id} is already stored in the database'.format(
                    **locals()))

        try:
            del kwargs['_id']
        except KeyError:
            pass

        graph = cls._build_extractor(_id, **kwargs)
        graph.remove_dead_nodes(iter(list(cls.features.values())))

        graph.process(**kwargs)
        return _id class NoPersistenceSettingsError(Exception):
    """
    Error raised when a BaseModel-derived class is used without an accompanying
    PersistenceSettings sub-class.
    """
    pass class ModelExistsError(Exception):
    """
    Error raised when a model has already been computed and persisted
    """
    pass class Feature(object):
    def __init__(
            self,
            extractor,
            needs=None,
            store=False,
            encoder=None,
            decoder=None,
            key=None,
            data_writer=None,
            persistence=None,
            **extractor_args):

        super(Feature, self).__init__()
        self.key = key
        self.extractor = extractor
        self.store = store
        self.encoder = encoder or IdentityEncoder

        self.needs = needs

        self.decoder = decoder or Decoder()
        self.extractor_args = extractor_args

        self._handle_callable_extractor()

        self.persistence = persistence

        if data_writer:
            self._data_writer = data_writer
        else:
            self._data_writer = DataWriter

    def aspect(self, aspect_key):
        return Aspect(aspect_key, self)

    def _fixup_needs(self):
        self.needs = dictify(self.needs, lambda item: item.key)

        for k, v in list(self.needs.items()):
            try:
                # the value is an Aspect
                self.needs[k] = v.feature

                class AspectExtractor(KeySelector, self.extractor):
                    def __init__(self, *args, **kwargs):
                        super(AspectExtractor, self).__init__(*args, **kwargs)

                self.extractor = AspectExtractor
                self.extractor_args.update(aspect_key=v.aspect_key)
            except AttributeError:
                # the value is already just a feature
                pass

    def _handle_callable_extractor(self):
        if inspect.isclass(self.extractor) \
                and issubclass(self.extractor, Node):
            return

        if callable(self.extractor):
            self.extractor_args = dict(
                func=self.extractor, **self.extractor_args)
            self.extractor = FunctionalNode
            return

        raise ValueError('extractor must be either a Node or a callable')

    def __repr__(self):
        return '{cls}(key = {key}, store = {store})'.format(
            cls=self.__class__.__name__, **self.__dict__)

    def __str__(self):
        return self.__repr__()

    @property
    def dependencies(self):
        return list(self.needs.values())

    @property
    def version(self):
        # KLUDGE: Build a shallow version of the extractor.  Building a deep
        # version with re-usable code is more difficult, because
        # self._build_extractor relies on this version property, so there's
        # a circular dependency.
        if isinstance(self.needs, dict):
            dependencies = dict(
                (k, v.extractor(**v.extractor_args))
                for k, v in list(self.needs.items()))
        else:
            dependencies = [f.extractor(**f.extractor_args) for f in self.needs]

        e = self.extractor(needs=dependencies, **self.extractor_args)
        return e.version

    def copy(
            self,
            extractor=None,
            needs=None,
            store=None,
            data_writer=None,
            persistence=None,
            extractor_args=None):
        """
        Use self as a template to build a new feature, replacing
        values in kwargs
        """
        f = Feature(
            extractor or self.extractor,
            needs=needs,
            store=self.store if store is None else store,
            encoder=self.encoder,
            decoder=self.decoder,
            key=self.key,
            data_writer=data_writer,
            persistence=persistence,
            **(extractor_args or self.extractor_args))
        f._fixup_needs()
        return f

    def database(self, persistence):
        return (self.persistence or persistence).database

    def keybuilder(self, persistence):
        return (self.persistence or persistence).key_builder

    def event_log(self, persistence):
        try:
            return (self.persistence or persistence).event_log
        except AttributeError:
            return None

    def feature_key(self, _id, persistence):
        return self.keybuilder(persistence).build(_id, self.key, self.version)

    def reader(self, _id, key, persistence):
        key = self.keybuilder(persistence).build(_id, key, self.version)
        return self.database(persistence).read_stream(key)

    @property
    def is_root(self):
        return not self.needs

    def _stored(self, _id, persistence):
        key = self.keybuilder(persistence).build(_id, self.key, self.version)
        return key in self.database(persistence)

    @property
    def content_type(self):
        return self.encoder.content_type

    def _can_compute(self, _id, persistence):
        """
        Return true if this feature stored, or is unstored, but can be computed
        from stored dependencies
        """
        if self.store and self._stored(_id, persistence):
            return True

        if self.is_root:
            return False

        return all(
            [n._can_compute(_id, persistence) for n in self.dependencies])

    def _compute(self, _id=None, persistence=None):
        graph, stream = self._build_partial(_id, persistence)

        kwargs = dict()
        for k, extractor in list(graph.roots().items()):
            try:
                kwargs[k] = extractor._reader
            except AttributeError:
                kwargs[k] = self.reader(_id, k, persistence)

        graph.process(**kwargs)
        return stream

    def compute(self, _id, persistence):
        self._compute(_id, persistence)

    def __call__(self, _id=None, decoder=None, persistence=None):
        if decoder is None:
            decoder = self.decoder

        try:
            raw = self.reader(_id, self.key, persistence)
            decoded = decoder(raw)
            return decoded
        except KeyError:
            pass

        if not self._can_compute(_id, persistence):
            raise AttributeError('%s cannot be computed' % self.key)

        stream = self._compute(_id, persistence)

        if stream is None:
            stream = self.reader(_id, self.key, persistence)

        stream.seek(0)
        decoded = decoder(stream)
        return decoded

    def _build_partial(self, _id, persistence):
        features = self._partial(_id, persistence=persistence)
        g = Graph()
        for feat in list(features.values()):
            e = feat._build_extractor(_id, g, persistence)
            if feat.key == self.key:
                stream = e.find_listener(
                    lambda x: isinstance(x, BytesIODataWriter))
                if stream is not None:
                    stream = stream._stream

        return g, stream

    def _partial(self, _id, features=None, persistence=None):
        """
        TODO: _partial is a shit name for this, kind of.  I'm building a graph
        such that I can only do work necessary to compute self, and no more
        """

        root = features is None

        stored = self._stored(_id, persistence)
        is_cached = self.store and stored

        if self.store and not stored:
            data_writer = None
        elif root:
            data_writer = BytesIODataWriter
        else:
            data_writer = None

        should_store = self.store and not stored
        nf = self.copy(
            extractor=DecoderNode if is_cached else self.extractor,
            store=root or should_store,
            needs=dict(),
            data_writer=data_writer,
            persistence=self.persistence,
            extractor_args=dict(decodifier=self.decoder, version=self.version) \
                if is_cached else self.extractor_args)

        if root:
            features = dict()

        features[self.key] = nf

        if not is_cached:
            for k, v in list(self.needs.items()):
                v._partial(_id, features=features, persistence=persistence)
                nf.needs[k] = features[v.key]

        return features

    def _depends_on(self, _id, graph, persistence, **kwargs):
        needs = OrderedDict()

        for k, f in list(self.needs.items()):
            if f.key in graph:
                needs[k] = graph[f.key]
                continue

            e = f._build_extractor(_id, graph, persistence, **kwargs)
            needs[k] = e

        return needs

    def _build_extractor(self, _id, graph, persistence, **kwargs):
        try:
            return graph[self.key]
        except KeyError:
            pass

        needs = self._depends_on(_id, graph, persistence, **kwargs)

        extractor_args = dict()
        for k, v in list(self.extractor_args.items()):
            if isinstance(v, Var) and v.name == k:
                try:
                    extractor_args[k] = kwargs[k]
                except KeyError:
                    raise ValueError('{k} is a Var, but it was not provided'
                                     .format(**locals()))
            else:
                extractor_args[k] = v

        e = self.extractor(needs=needs, **extractor_args)

        if isinstance(e, DecoderNode):
            reader = self.reader(_id, self.key, persistence)
            setattr(e, '_reader', reader)

        graph[self.key] = e

        if not self.store:
            return e

        key = self.key
        encoder = self.encoder(needs=e)
        graph['{key}_encoder'.format(**locals())] = encoder

        dw = self._data_writer(
            needs=encoder,
            _id=_id,
            feature_name=self.key,
            feature_version=self.version,
            key_builder=self.keybuilder(persistence),
            database=self.database(persistence),
            event_log=self.event_log(persistence))

        graph['{key}_writer'.format(**locals())] = dw
        return e class JSONFeature(Feature):
    def __init__(
            self,
            extractor,
            needs=None,
            store=False,
            key=None,
            encoder=JSONEncoder,
            data_writer=None,
            **extractor_args):
        super(JSONFeature, self).__init__(
            extractor,
            needs=needs,
            store=store,
            encoder=encoder,
            decoder=JSONDecoder(),
            key=key,
            data_writer=data_writer,
            **extractor_args) class CompressedFeature(Feature):
    def __init__(
            self,
            extractor,
            needs=None,
            store=False,
            key=None,
            **extractor_args):
        super(CompressedFeature, self).__init__(
            extractor,
            needs=needs,
            store=store,
            encoder=BZ2Encoder,
            decoder=BZ2Decoder(),
            key=key,
            **extractor_args) class ClobberJSONFeature(JSONFeature):
    def __init__(
            self,
            extractor,
            needs=None,
            store=False,
            key=None,
            encoder=JSONEncoder,
            **extractor_args):
        super(ClobberJSONFeature, self).__init__(
            extractor,
            needs=needs,
            store=store,
            encoder=encoder,
            key=key,
            data_writer=ClobberDataWriter,
            **extractor_args) class ClobberPickleFeature(PickleFeature):
    def __init__(
            self,
            extractor,
            needs=None,
            store=False,
            key=None,
            **extractor_args):
        super(ClobberPickleFeature, self).__init__(
            extractor,
            needs=needs,
            store=store,
            key=key,
            data_writer=ClobberDataWriter,
            **extractor_args) class TextFeature(Feature):
    def __init__(
            self,
            extractor,
            needs=None,
            store=False,
            key=None,
            **extractor_args):
        super(TextFeature, self).__init__(
            extractor,
            needs=needs,
            store=store,
            encoder=TextEncoder,
            decoder=GreedyTextDecoder(),
            key=key,
            **extractor_args) class ByteStream(Node):
    def __init__(self, chunksize=4096, needs=None):
        super(ByteStream, self).__init__(needs=needs)
        self._chunksize = int(chunksize)

    def _generator(self, stream, content_length):
        if not content_length:
            raise ValueError('content_length should be greater than zero')
        for chunk in chunked(stream, chunksize=self._chunksize):
            yield BytesWithTotalLength(chunk, content_length)

    def _from_http_response(self, resp):
        resp.raise_for_status()
        content_length = int(resp.headers['Content-Length'])
        return self._generator(resp.raw, content_length)

    def _handle_simple_get(self, data):
        parsed = urlparse(data)
        if parsed.scheme and parsed.netloc:
            resp = requests.get(data, stream=True)
            return self._from_http_response(resp)
        else:
            raise ValueError

    def _handle_http_request(self, data):
        s = requests.Session()
        prepped = data.prepare()
        resp = s.send(prepped, stream=True)
        return self._from_http_response(resp)

    def _handle_file_like_object(self, data):
        content_length = data.seek(0, 2)
        data.seek(0)
        return self._generator(data, content_length)

    def _handle_zip_file(self, data):
        return self._generator(data.zipfile, data.file_size)

    def _handle_file(self, data):
        with open(data, 'rb') as f:
            content_length = int(os.path.getsize(data))
            for chunk in self._generator(f, content_length):
                yield chunk

    def _get_strategy(self, data):
        if isinstance(data, ZipWrapper):
            return self._handle_zip_file
        if isinstance(data, requests.Request):
            return self._handle_http_request
        if isinstance(data, str):
            parsed = urlparse(data)
            is_url = parsed.netloc and parsed.scheme
            return self._handle_simple_get if is_url else self._handle_file
        return self._handle_file_like_object

    def _process(self, data):
        try:
            data = data.uri
        except AttributeError:
            pass
        strategy = self._get_strategy(data)
        for chunk in strategy(data):
            yield chunk class ByteStreamFeature(Feature):
    def __init__(
            self,
            extractor,
            needs=None,
            store=False,
            key=None,
            **extractor_args):
        super(ByteStreamFeature, self).__init__(
            extractor,
            needs=needs,
            store=store,
            encoder=BytesWithTotalLengthEncoder,
            decoder=BytesWithTotalLengthDecoder(
                chunksize=extractor_args['chunksize']),
            key=key,
            **extractor_args) def chunked(f, chunksize=4096):
    data = f.read(chunksize)
    while data:
        yield data
        data = f.read(chunksize) class LmdbDatabase(Database):
    def __init__(self, path, map_size=1000000000, key_builder=None):
        super(LmdbDatabase, self).__init__(key_builder=key_builder)
        self.path = path
        self.env = lmdb.open(
            self.path,
            max_dbs=10,
            map_size=map_size,
            writemap=True,
            map_async=True,
            metasync=True)
        self.env.reader_check()
        self.dbs = dict()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def close(self):
        self.env.close()

    def __del__(self):
        self.close()

    def _init_db_cache(self):
        with self.env.begin() as txn:
            cursor = txn.cursor()
            for feature in cursor.iternext(keys=True, values=False):
                self.dbs[feature] = self.env.open_db(feature)

    def _get_db(self, key):
        _id, feature, version = self.key_builder.decompose(from_bytes(key))
        versioned_key = self.key_builder.build(_id, version)
        feature = to_bytes(feature)
        try:
            return versioned_key, self.dbs[feature]
        except KeyError:
            db = self.env.open_db(feature)
            self.dbs[feature] = db
            return versioned_key, db

    def _get_read_db(self, key):
        _id, feature, version = self.key_builder.decompose(from_bytes(key))
        versioned_key = self.key_builder.build(_id, version)
        feature = to_bytes(feature)
        try:
            return versioned_key, self.dbs[feature]
        except KeyError:
            try:
                db = self.env.open_db(feature, create=False)
                self.dbs[feature] = db
                return versioned_key, self.dbs[feature]
            except lmdb.NotFoundError:
                raise KeyError(key)

    def write_stream(self, key, content_type):
        return WriteStream(key.encode(), self.env, self._get_db)

    def read_stream(self, key):
        _id, db = self._get_read_db(to_bytes(key))

        with self.env.begin(buffers=True) as txn:
            buf = txn.get(to_bytes(_id), db=db)
            if buf is None:
                raise KeyError(key)

        # POSSIBLE BUG:  Is it safe to keep the buffer around after the
        # transaction is complete?
        return BytesIO(buf)

    def size(self, key):
        _id, db = self._get_read_db(key)
        l = None

        with self.env.begin(buffers=True) as txn:
            buf = txn.get(_id.encode(), db=db)
            if buf is None:
                raise KeyError(key)
            l = len(buf)

        return l

    def _get_any_db(self):

        try:
            # return the first feature database
            # KLUDGE: This makes the assumption that features are never sparse,
            # i.e., all documents/ids have the same set of features
            return list(self.dbs.values())[0]
        except IndexError:
            pass

        # maybe another process has written data
        self._init_db_cache()

        try:
            return list(self.dbs.values())[0]
        except IndexError:
            return None

    def iter_ids(self):
        db = self._get_any_db()
        if db is None:
            return

        seen = set()
        with self.env.begin(db) as txn:
            cursor = txn.cursor(db)
            for _id in cursor.iternext(keys=True, values=False):
                _id, version = self.key_builder.decompose(_id.decode('utf-8'))
                if _id in seen:
                    continue
                yield _id
                seen.add(_id)

    def __contains__(self, key):
        try:
            _id, db = self._get_read_db(key)
        except KeyError:
            return False
        with self.env.begin(buffers=True) as txn:
            buf = txn.get(to_bytes(_id), db=db)
        return buf is not None

    def __delitem__(self, key):
        try:
            _id, db = self._get_read_db(key)
        except KeyError:
            return
        with self.env.begin(write=True) as txn:
            txn.delete(_id, db=db) class Decoder(object):
    """
    The simplest possible decoder takes a file-like object and returns it
    """

    def __init__(self):
        super(Decoder, self).__init__()

    def __call__(self, flo):
        return flo

    def __iter__(self, flo):
        for chunk in chunked(flo):
            yield chunk class PersistenceSettings(object):
    id_provider = UuidProvider()
    key_builder = StringDelimitedKeyBuilder()
    database = InMemoryDatabase(key_builder=key_builder)

    @classmethod
    def clone(cls, id_provider=None, key_builder=None, database=None):
        ip = id_provider
        kb = key_builder
        db = database

        class Settings(PersistenceSettings):
            id_provider = ip or cls.id_provider
            key_builder = kb or cls.key_builder
            database = db or cls.database

        return Settings
import requests
import http.client
import unittest2
import random
import subprocess
import sys
import time
import traceback
from .util import wait_for_http_server
from collections import defaultdict
from requests.exceptions import HTTPError
from .var import Var
from .extractor import NotEnoughData, Aggregator, Node, InvalidProcessMethod
from .iteratornode import IteratorNode
from .model import BaseModel, NoPersistenceSettingsError, ModelExistsError
from .feature import \
    Feature, JSONFeature, CompressedFeature, ClobberJSONFeature, \
    ClobberPickleFeature, TextFeature
from .data import *
from .bytestream import ByteStream, ByteStreamFeature
from io import BytesIO
from .util import chunked
from .lmdbstore import LmdbDatabase
from .decoder import Decoder
from .persistence import PersistenceSettings
from tempfile import mkdtemp
from shutil import rmtree
        super(SumUp, self).__init__(needs=needs)
        self._cache = dict()

    def _enqueue(self, data, pusher):
        self._cache[id(pusher)] = data

    def _dequeue(self):
        if len(self._cache) != len(self._needs):
            raise NotEnoughData()
        v = self._cache
        self._cache = dict()
        return v

    def _process(self, data):
        results = [str(sum(x)) for x in zip(*iter(data.values()))]
        yield ''.join(results)


class EagerConcatenate(Node):
    def __init__(self, needs=None):
        super(EagerConcatenate, self).__init__(needs=needs)
        self._cache = dict()

    def _enqueue(self, data, pusher):
        self._cache[id(pusher)] = data

    def _dequeue(self):
        v, self._cache = self._cache, dict()
        return v

    def _process(self, data):
        yield b''.join(iter(data.values()))


class NumberStream(Node):
    def __init__(self, needs=None):
        super(NumberStream, self).__init__(needs=needs)

    def _process(self, data):
        l = data_source[data]
        for i in range(0, len(l), 3):
            yield l[i: i + 3]


class Add(Node):
    def __init__(self, rhs=1, needs=None):
        super(Add, self).__init__(needs=needs)
        self._rhs = rhs

    def _process(self, data):
        yield [c + self._rhs for c in data]


class Tokenizer(Node):
    def __init__(self, needs=None):
        super(Tokenizer, self).__init__(needs=needs)
        self._cache = b''

    def _enqueue(self, data, pusher):
        self._cache += data

    def _finalize(self, pusher):
        self._cache += b' '

    def _dequeue(self):
        last_index = self._cache.rfind(b' ')
        if last_index == -1:
            raise NotEnoughData()
        current = self._cache[:last_index + 1]
        self._cache = self._cache[last_index + 1:]
        return current

    def _process(self, data):
        yield [x for x in data.split(b' ') if x]


class WordCount(Aggregator, Node):
    def __init__(self, needs=None):
        super(WordCount, self).__init__(needs=needs)
        self._cache = defaultdict(int)

    def _enqueue(self, data, pusher):
        for word in data:
            self._cache[word.lower().decode()] += 1


class IncrementalWordCount(Node):
    def __init__(self, needs=None, raise_for_chunk=None):
        super(IncrementalWordCount, self).__init__(needs=needs)
        self.count = defaultdict(int)
        self.raise_for_chunk = raise_for_chunk
        self._chunk = 0

    def _process(self, data):
        if self._chunk == self.raise_for_chunk:
            raise Exception('Raising for chunk {c}'.format(c=self._chunk))

        for word in data:
            self.count[word.lower().decode()] += 1
        yield self.count
        self._chunk += 1


class FeatureAggregator(Node):
    def __init__(self, cls=None, feature=None, needs=None):
        super(FeatureAggregator, self).__init__(needs=needs)
        self._cls = cls
        self._feature = feature

    def _process(self, data):
        db = data
        for key in db.iter_ids():
            try:
                doc = self._cls(key)
                yield getattr(doc, self._feature.key)
            except:
                pass


# TODO: Move these classes into the tests, where possible

class Document(BaseModel):