Execute class: $(command) in python

Although you can easily do with module subprocess what I do with the following class:

  • If you have Python 2.3 you may not have subprocess
  • This class is much simpler than subprocess: it does just one thing.
  • You may want to pipe two commands in a fancier way

I wrote this because I needed to use Python to do some system automation and I needed something that acted like Perl and bash `` (in bash 2 you are supposed to do $(command) instead of `command` but it’s syntactical sugar).

In my strive to be compatible with standard MacOS system (which du use 2.3 by default) I forgot it existed module subprocess. So I wrote this simple class.

Even if in production you may prefer to use subprocess you can consider this an example of streams in Python. This is a very simple example. If you want to really pipe two processes, use module subprocess. If you want to make streams in a C++ fashion you can take inspiration by this code, but you may want to do it line based (but this ain’t no problem, in fact we are using properties, so you can do whatever you want with them — an example later).

Stream acts like a Mixin (thanks Dialtone!). It defines __rshift__ and __lshift__ (that become >> and


class Stream(object):
    def __rshift__(self, rho):
        if isinstance(rho, Stream):
            rho.input = self.out
            return rho
        else: 
            raise TypeError()

    def __lshift__(self, rho):
        if isinstance(rho, Stream):
            self.input = rho.out
            return self
        else: raise TypeError()
    

Remember: classes that subclass Stream must have an attribute named out and one named input. If you need something more complex, use properties to implicitly call functions every-time you access out or in. For example since I use lazy evaluation I do:


    input = property(fset=set_input, fget=get_input)
    out = property(fget=get_out)
    err = property(fget=get_err)

Now have a look at the whole code:


import os

class Stream(object):
    def __rshift__(self, rho):
        if isinstance(rho, Stream):
            rho.input = self.out
            return rho
        else: 
            raise TypeError()

    def __lshift__(self, rho):
        if isinstance(rho, Stream):
            self.input = rho.out
            return self
        else: raise TypeError()

class Execute(Stream):
    def __init__(self, command, *kargs, **prefs):
        self._commandString = ' '.join((command,) + kargs)
        self.has_run = False

    def _lazy_run(self):
        if self.has_run: return
        self.has_run = True
        (self.sin, self.sout, self.serr) = os.popen3(self._commandString)
        self.sin.write(self.input)
        self.sin.close()        
        self.out_list = list([line.strip() for line in self.sout])
        self.err_list = list([line.strip() for line in self.serr])

    def run(self):
        self._lazy_run()
        return self

    def _make_string(self, which):
        self._lazy_run()
        if hasattr(self, "%s_string" % which):
            return
        setattr(self, "%s_string" % which, 
                    '\n'.join(getattr(self, "%s_list"% which)))

    def set_input(self, s):
        if hasattr(self, "input"):
            self.input_ = "%s%s" % (self.input, str(s))
        else:
            self.input_ = str(s)

    def get_input(self):
        try:
            return self.input_
        except AttributeError:
            return ""

    def get_out(self):
        self._make_string("out")
        return self.out_string

    def get_err(self):
        self._make_string("err")
        return self.err_string

    input = property(fset=set_input, fget=get_input)
    out = property(fget=get_out)
    err = property(fget=get_err)

    def __str__(self):
        return self.out

print Execute("cat Execute.py") >> Execute("wc -l")
print Execute("wc -l") << Execute("cat Execute.py") 

If you need to do different things, redefine (get|set)_(out|input|err).

Advertisements

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out / Change )

Twitter picture

You are commenting using your Twitter account. Log Out / Change )

Facebook photo

You are commenting using your Facebook account. Log Out / Change )

Google+ photo

You are commenting using your Google+ account. Log Out / Change )

Connecting to %s

%d bloggers like this: