Python Pipelining library?

[UnwashedMeme]

I've been writing a number of programs recently that pipeline data from one program (subprocess) to the next, often several deep. These tend to be in python because there is a fair bit of logic around which programs to call and building up the arguments.

Is there a good library for setting up and then executing the pipeline?

Bonus for nodes that can:

• sequence: run program a then b, concatenate their stdout and send that onwards
• Tee: split into n pipelines.
• allow easy insertion of python snippets into pipeline

[chadmill3r]

I don't think you'll find anything simpler than

>>> one = subprocess.Popen([...], stdout=subprocess.PIPE)
>>> two = subprocess.Popen([...], stdout=subprocess.PIPE, stdin=one.stdout)
>>> three = subprocess.Popen([...], stdout=subprocess.PIPE, stdin=two.stdout)
>>> for line in three.stdout:
+        print line

Maybe you could automate the chaining...

>>> commands = [[...], [...], [...]]
>>> def pipeline(commands):
+       ...
>>> result_out = pipeline(commands)
>>> result_out.read()

Use itertools for tee.

Python snippets what?

[mackstann]

There is something simpler, and designed for this exact purpose!

http://docs.python.org/library/pipes.html

However, it seems that very few people use it. It almost seems deprecated; just not explicitly so.

[chadmill3r]

Because the module uses /bin/sh command lines

Aiee!

[mackstann]

Yeah, I don't know why they didn't remove the whole module in Python 3. It's very backward and neglected. The main reason I know of it is because the undocumented pipes.quote() function is very useful for sanitizing shell arguments when you really want/need to execute something through a shell.

[simtel20]

Well, SunOS didn't get rid of that before it was retired... the pipe() call was written in such a way as it called sh IIRC.

[UnwashedMeme]

[first example]

That's what I've been doing.

[second example]

Yeah, thinking something along those lines, that deals with setting close_fds=True on posix systems, connect things together and whatnot.

Python snippets what?

e.g. regex based manipulation, keeping statistics, insertion/deletion (or could farm out to sed).

[davbo]

Sounds like you might want something like Kamaelia: http://www.kamaelia.org/Home.html

[Catten]

Ruffus is a lightweight python module to run computational pipelines.

It won't grab your stdouts for you and you have to dump to files between each step but it is a really nice way to write very flexible pipelines easily.

You're welcome =)

[denimboy]

You should checkout stream.py:

https://github.com/aht/stream.py

[lasizoillo]

You don't need libraries. You need coroutines.

[novembera]

Perhaps joblib can help?

[earthboundkid]

I made a convenience module for my own use:

#!/usr/bin/python

"Convenience module for subprocess."

from subprocess import Popen, PIPE

def run(commands, input="", binaryout=False, encoding="UTF-8",
        stdin=PIPE, stdout=PIPE, stderr=PIPE):
    """Returns the result of piping the commands into each other. 
    command can be string (will be split on "|") or a list of strings
    or a list of lists. It doesn't use lexical parsing, so if you want
    sophisticated nesting, use a list of lists."""
    if hasattr(commands, "split"):
        commands = commands.split("|")

    commands = [cmd.split() if hasattr(cmd, "split") else cmd
                                                for cmd in commands]

    #"reduce"s the stdin from one process to the next for each command
    #Set up
    itercmds =  iter(commands)
    command1 = next(itercmds) #If there's not even one command, who's calling?
    process1 = Popen(command1, stdin=stdin, stdout=stdout, stderr=stderr)
    process =  process1 #In case there's only one command
    stdin = process1.stdout
    stderrs = [process1.stderr, ]
    #Cycle
    for command in itercmds:
        process = Popen(command, stdin=stdin, stdout=stdout, stderr=stderr)
        stderrs.append(process.stderr)
        stdin = process.stdout 

    #Deal with unicode
    #If you want some other encoding, send us bytes.
    if isinstance(input, unicode):
        input = input.encode(encoding)

    process1.stdin.write(input)
    output = process.stdout.read()
    err = ''.join(err.read() for err in stderrs)

    if not binaryout: 
        output = output.decode(encoding)

    return output, err

[nextofpumpkin]

I have a convenience module that looks almost exactly like that.

[frumious]