Skip to main content
https://imgur.com/Q9Lv0xo.png
https://travis-ci.org/dokipen/tubing.svg?branch=master https://coveralls.io/repos/github/dokipen/tubing/badge.svg?branch=master https://img.shields.io/pypi/v/tubing.svg https://img.shields.io/pypi/pyversions/tubing.svg https://img.shields.io/pypi/dd/tubing.svg https://img.shields.io/pypi/l/tubing.svg https://img.shields.io/pypi/wheel/tubing.svg https://readthedocs.org/projects/tubing/badge/?version=latest Code Health

Tubing is a Python I/O library. What makes tubing so freakin’ cool is the gross abuse of the bit-wise OR operator (|). Have you ever been writing python code and thought to yourself, “Man, this is great, but I really wish it was a little more like bash.” Whelp, we’ve made python a little more like bash.If you are a super lame nerd-kid, you can replace any of the bit-wise ORs with the tube() function and pray we don’t overload any other operators in future versions. Here’s how you install tubing:

$ pip install tubing

Tubing is pretty bare-bones at the moment. We’ve tried to make it easy to add your own functionality. Hopefully you find it not all that unpleasant. There are three sections below for adding sources, tubes and sink. If you do make some additions, think about committing them back upstream. We’d love to have a full suite of tools.

Now, witness the power of this fully operational I/O library.

from tubing import sources, tubes, sinks

objs = [
    dict(
        name="Bob Corsaro",
        birthdate="08/03/1977",
        alignment="evil",
    ),
    dict(
        name="Tom Brady",
        birthdate="08/03/1977",
        alignment="good",
    ),
]
sources.Objects(objs) \
     | tubes.JSONDumps() \
     | tubes.Joined(by=b"\n") \
     | tubes.Gzip() \
     | sinks.File("output.gz", "wb")

Then in our old friend bash.

$ zcat output.gz
{"alignment": "evil", "birthdate": "08/03/1977", "name": "Bob Corsaro"}
{"alignment": "good", "birthdate": "08/03/1977", "name": "Tom Brady"}
$

You can find more documentation on readthedocs

Catalog

Sources

Objects

Takes a list of python objects.

File

Creates a stream from a file.

Bytes

Takes a byte string.

IO

Takes an object with a read function.

Socket

Takes an addr, port and socket() args. .

HTTP

Takes an method, url and any args that can be passed to requests library.

Tubes

Gunzip

Unzips a binary stream.

Gzip

Zips a binary stream.

JSONLoads

Parses a byte string stream of raw JSON objects. Will try to use ujson, then built-in json.

JSONDumps

Serializes an object stream using json.dumps. Will try to use ujson, then built-in json.

Split

Splits a stream that supports the split method.

Joined

Joins a stream of the same type as the by argument.

Tee

Takes a sink and passes chunks along apparatus.

Map

Takes a transformer function for single items in stream.

Filter

Takes a filter test callback and only forwards items that pass.

ChunkMap

Takes a transformer function for batch of stream items.

Sinks

Objects

A list that stores all passed items to self.

Bytes

Saves each chunk self.results.

File

Writes each chunk to a file.

HTTPPost

Writes data via HTTPPost.

Hash

Takes algorithm name, updates hash with contents.

Debugger

Writes each chunk to the tubing.tubes debugger with level DEBUG.

Extensions

s3.S3Source

Create stream from an S3 object.

s3.MultipartUploader

Stream data to S3 object.

elasticsearch.BulkSink

Stream elasticsearch.DocUpdate objects to the elasticsearch _bulk endpoint.

Sources

To make your own source, create a Reader class with the following interface.

class MyReader(object):
    """
    MyReader returns count instances of data.
    """
    def __init__(self, data="hello world\n", count=10):
        self.data = data
        self.count = count

    def read(self, amt):
        """
        read(amt) returns $amt of data and a boolean indicating EOF.
        """
        if not amt:
            amt = self.count
        r = self.data * min(amt, self.count)
        self.count -= amt
        return r, self.count <= 0

The important thing to remember is that your read function should return an iterable of units of data, not a single piece of data. Then wrap your reader in the loving embrace of MakeSourceFactory.

from tubing import sources

MySource = sources.MakeSourceFactory(MyReader)

Now it can be used in a apparatus!

from __future__ import print_function

from tubing import tubes
sink = MySource(data="goodbye cruel world!", count=1) \
     | tubes.Joined(by=b"\n") \
     | sinks.Bytes()

print(sinks.result)
# Output: goodbye cruel world!

Tubes

Making your own tube is a lot more fun, trust me. First make a Transformer.

class OptimusPrime(object):
    def transform(self, chunk):
        return list(reversed(chunk))

chunk is an iterable with a len() of whatever type of data the stream is working with. In Transformers, you don’t need to worry about buffer size or closing or exception, just transform an iterable to another iterable. There are lots of examples in tubes.py.

Next give Optimus Prime a hug.

from tubing import tubes

AllMixedUp = tubes.MakeTranformerTubeFactory(OptimusPrime)

Ready to mix up some data?

from __future__ import print_function

import json
from tubing import sources, sinks

objs = [{"number": i} for i in range(0, 10)]

sink = sources.Objects(objs) \
     | AllMixedUp(chunk_size=2) \
     | sinks.Objects()

print(json.dumps(sink))
# Output: [{"number": 1}, {"number": 0}, {"number": 3}, {"number": 2}, {"number": 5}, {"number": 4}, {"number": 7}, {"number": 6}, {"number": 9}, {"number": 8}]

Sinks

Really getting tired of making documentation… Maybe I’ll finish later. I have real work to do.

Well.. I’m this far, let’s just push through.

from __future__ import print_function
from tubing import sources, tubes, sinks

class StdoutWriter(object):
    def write(self, chunk):
        for part in chunk:
            print(part)

    def close(self):
        # this function is optional
        print("That's all folks!")

    def abort(self):
        # this is also optional
        print("Something terrible has occurred.")

Debugger = sinks.MakeSinkFactory(StdoutWriter)

objs = [{"number": i} for i in range(0, 10)]

sink = sources.Objects(objs) \
     | AllMixedUp(chunk_size=2) \
     | tubes.JSONDumps() \
     | tubes.Joined(by=b"\n") \
     | Debugger()
# Output:
#{"number": 1}
#{"number": 0}
#{"number": 3}
#{"number": 2}
#{"number": 5}
#{"number": 4}
#{"number": 7}
#{"number": 6}
#{"number": 9}
#{"number": 8}
#That's all folks!

Metadata

Release files for tubing 0.0.2.post170

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for tubing 0.0.2.post170
File Size Uploaded Provenance Actions
tubing-0.0.2.post170.tar.gz 17.0 kB Without provenance Details for tubing-0.0.2.post170.tar.gz

Built distribution (wheel)

Table of built distributions (wheels) for tubing 0.0.2.post170
File Interpreter ABI Platform Actions
tubing-0.0.2.post170-py2.py3-none-any.whl Python 3, Python 2 none any Details for tubing-0.0.2.post170-py2.py3-none-any.whl

Total release size: 38.5 kB

Release files / tubing-0.0.2.post170.tar.gz

Download URL tubing-0.0.2.post170.tar.gz
Size 17.0 kB
Tags Source
SHA-256 checksum
How to use checksums
e0fadddcf40a9cb63681a4dba9cd87b3e831dffe91904aef983f314e19511f42
BLAKE2b-256 checksum
How to use checksums
a7fe94734a57676a9f687b13327fbb06d8b742731c818a7b9f4b317b47f4be23
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No

Release files / tubing-0.0.2.post170-py2.py3-none-any.whl

Download URL tubing-0.0.2.post170-py2.py3-none-any.whl
Size 21.5 kB
Tags Python 2 Python 3
SHA-256 checksum
How to use checksums
8a315c58d65ca7bc61ca7c6bc600c021956eba317ceb081e0bef69284340b41b
BLAKE2b-256 checksum
How to use checksums
25f55fc267c0027cf702ce1ae3046cae27898c366825f949e77830d6c0fd396b
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No

Security for tubing 0.0.2.post170

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

No provenance found

None of the files in this release were uploaded with provenance metadata.

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.

To learn more about protecting yourself from malicious actors, visit our package security documentation.

Release history Release notifications | RSS feed

This release
Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page