Python - Subprocess

From XennisWiki
Jump to: navigation, search

With Python you can open a subprocess, e.g. call a command line program in another programming language.

Simple Popen example

def execute_cmd(args, cwd=None, shell=False):
    # type: (list, typing.Optional[str], typing.Optional[bool]) -> (typing.Optional[int], typing.Optional[str], typing.Optional[str])
    process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd, shell=False)
    (stdout, stderr) = process.communicate()
    return process.returncode, stdout, stderr


def execute_cmd_with_err_msg(args):
    # type: (list) -> None
    exit_code, _stdout, stderr = execute_cmd(args)
    if exit_code != 0:
        msg = 'exit code {} from cmd \'{}\''.format(exit_code, ' '.join(args))
        if stderr:
            msg += ', error:\n{}'.format(stderr)
        raise Exception(msg)


def command_output(args, shell=False):
    output, error = execute_cmd(args, shell=shell)
    if error:
        logging.warn("Error executing command %s: %s" % (cmd, error))
        return

   return output

Popen example

my_app.py (your application, which starts a process)

import subprocess32 as subprocess

class MyApp:
    
    def __init__(self):
        print('Hello, I am MyApp')
        self.do()
        
    def do(self):
        print('Start process -----------------------------------')
        process = self.external_process(["python", "my_process.py", "-s", "10"])
        print('exit_status -------------------------------------')
        print(process[0])
        print('output-------------------------------------------')
        print(process[1])
        print('error -------------------------------------------')        
        print(process[2])        
        
    def external_process(self, process_args, input_date='', timeout=None):      
        process = subprocess.Popen(process_args,
                                   stdout=subprocess.PIPE,
                                   stdin=subprocess.PIPE,
                                   stderr=subprocess.PIPE)
        try:
            (stdout, stderr) = process.communicate(input_date, timeout)
        except subprocess.TimeoutExpired as e:
            # cleanup process
            # see https://docs.python.org/3.3/library/subprocess.html?highlight=subprocess#subprocess.Popen.communicate
            process.kill()
            process.communicate()
            raise e

        exit_status = process.returncode
        return (exit_status, stdout, stderr)       
        
if __name__ == '__main__':
    # run the application
    MyApp()

my_process.py

import sys
import getopt

class MyProcess:
    
    def __init__(self):
        print('Hello, I am MyProcess')
    
    def do(self, start, end):
        for i in range(0, 10):
            print('start=%s, end=%s' % (start, end))

if __name__ == '__main__':
    try:
        opts, args = getopt.getopt(sys.argv[1:], "hs:e:", ["help", "start=", "end="])
    except getopt.GetoptError as e:
        print(str(e))
        print('Usage: my-process.py -s <start> -e <end>')
        print('Usage: my-process.py --start=<start> -end=<end>')
        sys.exit(2)    

    start = 0
    end = 10    
    for opt, arg in opts:
        if opt == '-h':
            print('Usage: my-process.py -s <start> -e <end>')
            sys.exit()
        elif opt in ("-s", "--start"):
            start = arg;
        elif opt in ("-e", "--end"):
            end = arg
        else:
            raise Exception("unhandled option")

    
    program = MyProcess()
    program.do(start, end)
    sys.exit()

See also