1. Troy Williams
  2. hg_utilities


hg_utilities / SubprocessHelper.py

#!/usr/bin/env python
#-*- coding:utf-8 -*-

Contains code that helps in dealing with the subprocess module.


Copyright (c) 2011 Troy Williams

License: The MIT License (http://www.opensource.org/licenses/mit-license.php)

__uuid__ = '4d6778c8-41e3-46a5-947c-d48ddeb36a50'
__author__ = 'Troy Williams'
__email__ = 'troy.williams@bluebill.net'
__copyright__ = 'Copyright (c) 2011, Troy Williams'

#Here is a quick method to determine the string tokenization from:
#>>> import shlex, subprocess
#>>> command_line = raw_input()
#/bin/vikings -input eggs.txt -output "spam spam.txt" -cmd "echo '$MONEY'"
#>>> args = shlex.split(command_line)
#>>> print args
#['/bin/vikings', '-input', 'eggs.txt', '-output', 'spam spam.txt', '-cmd', "echo '$MONEY'"]
#>>> p = subprocess.Popen(args) # Success!

import subprocess

class SubprocessHelper():
    A class that wraps calls to subprocess.Popen. It takes a command in the form
    of a list and a dictionary of arguments. The dictionary can contain the
    following values: shell and working_directory.
    shell - indicates to subpress that the shell should be used to interpret the
    command. Defaults to False

    working_directory - the path to the folder which will be the working folder.
    Defaults to None.

    It stores the results of the output stream and error stream to lists. if the
    command was successful, the standard_error_stream will be empty.

    def __init__(self):
        self.standard_output_stream = []
        self.standard_error_stream = []

    def run(self, command, **kwargs):
        Takes the list and attempts to run it in the command shell and waits for it
        to execute. It then returns the results to the caller.
        if not command:
            raise Exception, 'Valid command required - fill the list please!'

        p = subprocess.Popen(command,
                             shell=kwargs.get('shell', False),
                             cwd=kwargs.get('working_directory', None))

        output_stream, error_stream = p.communicate()

        if output_stream:
            self.standard_output_stream = output_stream.strip().split('\n')

        if error_stream:
            self.standard_error_stream = error_stream.strip().split('\n')