Execute class: $(command) in python
February 23rd, 2006
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).
Leave a Reply