Subprocess

Subprocess#

Instead of using dedicated task runners, one can also be in need of launching external commands directly from inside a Python script. The subprocess module is part of Python’s Standard Library and helps spawning new processes and managing their input/output.

Simple example#

Let’s consider the following shell command as an example,

$ seq 20 | grep 3
3
13

where seq 20 prints lines with the integers 1 to 20, | redirects this output as input to the next command, and finally grep 3 keeps only lines containing the digit 3.

We can use subprocess.run() to execute this command in Python and get its output.

import subprocess

some_cmd = "seq 20 | grep 3"
some_cmd_result = subprocess.run(some_cmd, shell=True, capture_output=True)

The shell=True argument executes the command in a shell (which on POSIX defaults to /bin/sh), whereas the capture_output=True is a short-hand to keep the output of stdout and stderr (instead of simply ignoring it). The latter is crucial if you want to parse the output of the command (and be able to handle any error messages sent to stderr). This data can be accessed as attributes of the returned object.

print(some_cmd_result)  # CompletedProcess(args='seq 20 | grep 3', returncode=0, stdout=b'3\n13\n', stderr=b'')
print(some_cmd_result.stdout.decode("utf-8"))  # '3\n13\n'
print(some_cmd_result.stdout.decode("utf-8").splitlines())  # ['3', '13']

Given that .stdout is a bytestring, we must first use .decode("utf-8") to convert it into a normal Python string. An additional .splitlines() then converts the string with newlines into a Python list.

Example wrapper function#

If you need to call subprocess.run() multiple times, the bytestring decoding and error handling may become repetitive. The following snippet is meant to simplify this by wrapping the required steps in a Python function.

import subprocess


def run_command(command, shell=True, timeout=3):
    """Run generic commands via subprocess wrapper and return their stdout as string."""
    try:
        cmd = subprocess.run(
            command, shell=shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout
        )
        if cmd.returncode != 0:
            raise Exception("command returned non-zero exit code: " + str(cmd.returncode))
        if cmd.stderr:
            stderr = cmd.stderr.decode("utf-8")
            raise Exception("command returned output to stderr: " + stderr)
        else:
            command_output = cmd.stdout.decode("utf-8")
    except subprocess.TimeoutExpired:
        raise Exception("command timeout")

    return command_output

Here an example usage of this wrapper function.

some_cmd_results_list = run_command("seq 20 | grep 3").splitlines()
Contents