Python subprocess

From wikinotes

If you want to run a command on the host operating system, you'll want to use python's subprocess module. It provides a means of safely spawning a subprocess, and passing parameters in a way that does not expose security vulnerabilities.

There are a lot of variations on how this can be used, these are only the methods I use most frequently.


SubProcesses

Basics

Popen Shortcuts

import subprocess

# shortcuts
subprocess.check_call(['netstat', '-an'], universal_newlines=True)

The Popen objects are the most powerful, but most verbose way of managing processes.

# redirecting streams between processes
pipe1 = subprocess.Popen(['tasklist'], stdout=subprocess.PIPE)
pipe2 = subprocess.Popen(['find', '/I', 'python'], stdin=pipe1.stdout, stdout=subprocess.PIPE)
(stdout, stderr) = pipe2.communicate()

Windows

Windows executables are either built as console-applications (require console) or gui-applications (do not show console).
On windows, you can use pythonw.exe to launch an application without a console.
Regardless of this decision, you can control whether or not a subprocess should open a new console while it runs.

import subprocess

# don't open new console window for subprocess
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
subprocess.Popen(['foo.exe'], startupinfo=startupinfo)

wrap stdin/stdout/stderr in subprocess

Wrap a subprocess, mirroring stdin/stdout/stderr from outside to inside


It's useful if you want to wrap an interactive process that requires input to stdin.

#!/usr/bin/env python

import enum
import os
import re
import select
import subprocess
import sys


class MiddlewareBase:
    """ Base Class for middleware
    """
    def apply(self, text):
        raise NotImplementedError()


class WrapSubprocess:
    def wrap(self, pipe, stdout_middleware=None, stderr_middleware=None):
        """ Wraps a subprocess.Popen(), allowing you to modify it's stdin/stdout/stderr with lists of 'middleware' functions.
        """
        while True:
            # exit when subprocess exits
            if pipe.poll() is not None:
                sys.exit(pipe.returncode)

            readable, _, _ = select.select([sys.stdin.fileno(), pipe.stdout.fileno(), pipe.stderr.fileno()], [], [])

            if pipe.stdout.fileno() in readable:
                self._read_stream_to(pipe.stdout, sys.stdout, stdout_middleware)

            if pipe.stderr.fileno() in readable:
                self._read_stream_to(pipe.stderr, sys.stderr, stderr_middleware)

            if sys.stdin.fileno() in readable:
                self._write_stream_to(sys.stdin.buffer, pipe.stdin)

    def _read_stream_to(self, input, output, middleware_stack=None):
        """ Repeats all bytes written to file-object input to file-object output.
        """
        text = input.peek()   # preview available bytes
        input.read(len(text)) # hacky seek
        if text:
            text = self._apply_middleware(middleware_stack, text.decode("utf-8"))
            output.write(text)
            output.flush()

    def _write_stream_to(self, input, output):
        """ Repeats all bytes read from file-object input to file-object output.
        """
        text = input.peek()   # preview available bytes
        input.read(len(text)) # hacky seek
        if isinstance(text, bytes):
            output.write(text)
        else:
            output.write(text.encode("utf-8"))
        output.flush()

    def _apply_middleware(self, middleware_stack, text):
        """ Calls apply(text) method on each `Middleware`, which returns a new text object.
        """
        if not middleware_stack:
            return text

        for middleware in middleware_stack:
            text = middleware.apply(text)
        return text


if __name__ == '__main__':
    wrapper = WrapSubprocess()

    cmds = ["rails", "test"] + sys.argv
    cmds.remove(__file__)
    pipe = subprocess.Popen(
        cmds,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env=os.environ
    )
    wrapper.wrap(
        pipe,
        stdout_middleware=middleware_stack,
        stderr_middleware=middleware_stack,
    )


Process Exitcodes

import sys

sys.exit(1)  # exit with fail

Environment Variables

import os

os.environ['FOO'] = 'bar'