Welcome to Tubing’s documentation!¶

Tubes¶

For most users, they simply want to transform elements of data as it goes through the stream. This can be achieved simply with the following idiom:

SomeSource | [Tubes ..] | tubes.Map(lambda x: transform(x)) | [Tubes ..] | SomeSink

Sometimes you’d like to transform an entire chunk of data at a time, instead of one element at a time:

SomeSource | [Tubes ..] | tubes.ChunkMap(lambda x: transform(x)) | [Tubes ..] | SomeSink

Other times you just want to filter out some data:

SomeSource | [Tubes ..] | tubes.Filter(lambda x: x > 10) | [Tubes ..] | SomeSink

All of these general tube tools also take close_fn and abort_fn params and are shorthand for creating your own Tube class.

Of course, if you need to keep state, you can create a closure, but at some point, that can become cumbersome. You might also want to make a reusable Tube, in that case it could be nice to make a

The easiest way to extend tubing is to create a Transformer, and use TransformerTubeFactory decorator to turn it into a Tube. A Transformer has the following interface:

@tubes.TransformerTubeFactory()
class NewTube(object):
    def transform(self, chunk):
        return new_chunk

    def close(self):
        return last_chunk or None

    def abort(self):
        pass

A chunk is an iterable of whatever type of stream we are working on, whether it be bytes, Unicode characters, strings or python objects. We can index it, slice it, or iterate over it. transform simple takes a chunk, and makes a new chunk out of it. TransformerTubeFactory will take care of all the dirty work. Transformers are enough for most tasks, but if you need to do something more complex, you may need to go deeper.

First let’s describe how tubes work in more detail. Here’s the Tube interface:

# tube factory can be a class

class TubeFactory(object):
    # This is what we export, and what is called when users create a tube.
    # The syntax looks like this:
    # SourceFactory() | [TubeFactory()...] | SinkFactory()
    def __call__(self, *args, **kwargs):
        return Tube()

#  or a function

def TubeFactory(*args, **kwargs):
    return Tube()

# ------------------------
class Tube(object):
    def receive(self, source):
        # return a TubeWorker
        tw = TubeWorker
        tw.source = source
        return tw

class TubeWorker(object):
    def tube(self, receiver):
        # receiver is they guy who will call our `read` method. Either
        # another Tube or a Sink.
        return receiver.receive(self)

    def __or__(self, *args, **kwargs):
        # Our reason for existing.
        return self.tube(*args, **kwargs)

    def read(self):
        # our receiver will call this guy. We return a tuple here of
        # `chunk, eof`.  We should return a chunk of len amt of whatever
        # type of object we produce. If we've exhausted our upstream
        # source, then we should return True as the second element of our
        # tuple. The chunk size should be configuratable and read should
        # return a len() of chunk size or less.
        return [], True

A TubeFactory is what casual users deal with. As you can see, it can be an object or a function, depending on your style. It’s easier for me to reason about state with an object, but if you prefer a closure, go for it! Classes are just closures with more verbose states, after all.

When a casual is setting up some tubing, the TubeFactory returns a Tube, but this isn’t the last object we’ll create. The Tube doesn’t have a source connected, so it’s sort of useless. It’s just a placeholder waiting for a source. As soon as it gets a source, it will hand off all of it’s duties to a TubeWorker.

A TubeWorker is ready to read from it’s source, but it doesn’t. TubeWorkers are pretty lazy and need someone else to tell them what to do. That’s where a receiver comes in to play. A receiver can be another Tube, or a Sink. If it’s another Tube, you know the drill. It’s just another lazy guy that will only tell his source to read when his boss tells him to read. Ultimately, the only guy who wants to do any work is the Sink. At the end of the chain, a sink’s receive function will be called, and he’ll get everyone to work.

Technically, we could split the TubeWorker interface into two parts, but it’s not really necessary since they share the same state. We could also combine TubeFactory, Tube and TubeWorker, and just build up state overtime. I’ve seriously consider this, but I don’t know, this feels better. I admit, it is a little complicated, but one advantage you get is that you can do something like this:

from tubing import tubes

tube = tubes.GZip(chunk_size=2**32)

source | tube | output1
source | tube | output2

Since tube is a factory and not an object, each tubeline will have it’s own state. tubeline.... I just made that up. That’s an execution pipeline in tubing. But we don’t want to call it a pipeline, it’s a tubeline. Maybe there’s a better name? I picture a chemistry lab with a bunch of transparent tubes connected to beakers and things like that.

Let’s call it an apparatus.

Sources¶

TODO

Sinks¶

TODO

Things You Can’t do with Tubing¶

• Tee to another apparatus
• async programming
• your laundry

Subpackages¶

Submodules¶

tubing.compat module¶

compat provides tools to make code compatible across python versions.

tubing.compat.python_2_unicode_compatible(klass)¶

lifted from Django A decorator that defines __unicode__ and __str__ methods under Python 2. Under Python 3 it does nothing.

To support Python 2 and 3 with a single code base, define a __str__ method returning text and apply this decorator to the class.

tubing.sinks module¶

class tubing.sinks.HTTPPost(url, username=None, password=None, chunks_per_post=1024, response_handler=<function <lambda>>)¶

Bases: object

HTTPPost doesn’t support the write method, and therefore can not be used with tubes.Tee.

gen()¶

We have to do this goofy shit because requests.post doesn’t give access to the socket directly. In order to stream, we need to pass a generator object to requests.

tubing.sources module¶

class tubing.sources.MakeSourceFactory(reader_cls, default_chunk_size=65536)¶

Bases: object

MakeSourceFactory takes a reader object and returns a Source factory.

class tubing.sources.Source(reader, chunk_size)¶

Bases: object

Source is a wrapper for Readers that allows piping.

tubing.tubes module¶

tubing.tubes.MakeTransformerTubeFactory(transformer_cls, default_chunk_size=262144)¶

Returns a TransformerTubeFactory, which in turn, returns a TransformerTube.

class tubing.tubes.TransformerTube(transformer_cls, default_chunk_size, *args, **kwargs)¶

Bases: object

TransformerTube is what is returned by a TransformerTubeFactory. It manages the initialization of the TransformerTubeWorker.

tubing.tubes.TransformerTubeFactory(default_chunk_size=262144)¶

TransformerTubeFactory is a decorator to turn a Transformer class into a TransformerTubeFactory.

class tubing.tubes.TransformerTubeWorker(apparatus, chunk_size, transformer)¶

Bases: object

TransformerTubeWorker wraps a Transformer and does all the grunt work that most tubes need to do. Transformers should implement transform(chunk), and optionally close() and abort().

append(chunk)¶

append to the buffer, creating it if it doesn’t exist.

buffer_len()¶

buffer_len even if buffer is None.

read()¶

This is where the rubber meets the snow.

read_complete()¶

read_complete tells us if the current request is fulfilled. It’s fulfilled if we’ve reached the EOF in the source, or we have $amt parts. If amt is None, we should read to the source’s EOF.

shift_buffer(amt)¶

Remove $amt data from the front of the buffer and return it.

class tubing.tubes.TubeIterator(tube)¶

Bases: object

TubeIterator wraps a tube in an iterator object.