#!/usr/bin/env python """ shell.py - Python Shell Commands Author: Sean B. Palmer, inamidst.com """ from subprocess import Popen, PIPE def chomp(line): if line.endswith('\n'): return line[:-1] class Process(object): def __init__(self, args): self.args = args self.input = None def __iter__(self): process = self pipeline = [] while True: pipeline = [process.args] + pipeline if process.input is None: break process = process.input first = pipeline[0] procs = [Popen(first, stdout=PIPE)] for args in pipeline[1:]: procs.append(Popen(args, stdin=procs[-1].stdout, stdout=PIPE)) stdout = procs[-1].stdout return iter(chomp(line) for line in stdout) def __or__(self, obj): if not isinstance(obj, Process): raise TypeError("Expected Process, got %r" % obj) obj.input = self return obj def out(*args): return Process(args) def do(seq): for item in seq: pass def run(*args): do(out(*args)) def test(): output = out("cat", "shell.py") | out("grep", ".") | out("wc", "-w") for line in output: print line print cp = lambda *args: out(*["cp"] + list(args)) ls = lambda *args: out(*["ls"] + list(args)) rm = lambda *args: out(*["rm"] + list(args)) do(cp("shell.py", "test.tmp")) for line in ls("-al", "test.tmp"): print line do(rm("test.tmp")) if __name__ == '__main__': test()