brief introduction

The subprocess module allows us to start a new process and connect to their input / output / error pipeline to get the return value.
Official documents of Python 3: https://docs.python.org/zh-cn/3/library/subprocess.html

use

The first recommended method for the subprocess module is its run method. For more advanced usage, you can directly use the Popen interface.

run() method

subprocess.run(args, *, stdin=None, input=None, stdout=None, stderr=None, capture_output=False, shell=False, cwd=None, timeout=None, check=False, encoding=None, errors=None, text=None, env=None, universal_newlines=None)

• args: used as a parameter to start the process It could be a list or a string.
• stdin, stdout, and stderr: Standard inputs, outputs, and errors for child processes. Its value can be subprocess PIPE,subprocess.DEVNULL, an existing file descriptor, an open file object, or none. subprocess.PIPE means to create a new pipe for the child process. subprocess.DEVNULL indicates OS devnull. None is used by default, which means nothing is done. In addition, stderr can be merged into stdout and output together.
• Timeout: sets the command timeout. If the command execution time expires, the child process will be killed and a timeout expired exception will pop up.
• check: if this parameter is set to True and the process exit status code is not 0, a CalledProcessError exception will pop up.
• Encoding: if this parameter is specified, stdin, stdout and stderr can receive string data and encode it in this encoding method. Otherwise, only bytes of data will be received.
• Shell: if this parameter is True, the specified command will be executed through the shell of the operating system.

The call method of run returns the CompletedProcess instance, which is similar to the direct Popen. The implementation is the same. In fact, Popen is called, which is roughly the same as the Popen constructor, for example:

#Execute the ls -l /dev/null command
>>> subprocess.run(["ls", "-l", "/dev/null"])
crw-rw-rw-  1 root  wheel    3,   2  5  4 13:34 /dev/null
CompletedProcess(args=['ls', '-l', '/dev/null'], returncode=0)

returncode: the status of the child process after execution. Generally, if the return status is 0, it indicates that it has been run. If the value is negative "- N", it indicates that the child process has been terminated.

import subprocess
def runcmd(command):
    ret = subprocess.run(command,shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE,encoding="utf-8",timeout=1)
    if ret.returncode == 0:
        print("success:",ret)
    else:
        print("error:",ret)


runcmd(["dir","/b"])#Sequence parameters
runcmd("exit 1")#String parameter

The output results are as follows:

success: CompletedProcess(args=['dir', '/b'], returncode=0, stdout='test.py\n', stderr='')
error: CompletedProcess(args='exit 1', returncode=1, stdout='', stderr='')

Popen() method

Pope is the core of the subprocess, which handles the creation and management of subprocesses.

Constructor:

class subprocess.Popen(args, bufsize=-1, executable=None, stdin=None, stdout=None, stderr=None, 
preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, universal_newlines=False, 
startupinfo=None, creationflags=0,restore_signals=True, start_new_session=False, pass_fds=(),
*, encoding=None, errors=None)

Common parameters:

• args: shell command, which can be string or sequence type (e.g. list, tuple)
• bufsize: buffer size. When creating a pipe object for standard flow, use - 1 by default.
  0: do not use buffer
  1: Indicates line buffering, only if universal_ Available when newlines = true, that is, text mode
  Positive number: indicates the buffer size
  Negative number: indicates that the system default buffer size is used.
• stdin, stdout, stderr: respectively represent the standard input, output and error handle of the program
• preexec_fn: only valid under Unix platform. It is used to specify an executable object that will be called before the child process runs
• Shell: if this parameter is True, the specified command will be executed through the shell of the operating system.
• cwd: used to set the current directory of child processes.
• Env: environment variable used to specify the child process. If env = None, the environment variables of the child process are inherited from the parent process.

Create a child process and execute a simple command:

>>> import subprocess
>>> p = subprocess.Popen('ls -l', shell=True)
>>> total 164
-rw-r--r--  1 root root   133 Jul  4 16:25 admin-openrc.sh
-rw-r--r--  1 root root   268 Jul 10 15:55 admin-openrc-v3.sh
...
>>> p.returncode
>>> p.wait()
0
>>> p.returncode

Popen object method

• poll(): check whether the process is terminated. If it is terminated, return code; otherwise, return None.
• wait(timeout): wait for the child process to terminate.
• communicate(input,timeout): interact with child processes, send and read data.
• send_ Signal (sinnal): Send a signal to the child process.
• terminate(): stop the child process, that is, send SIGTERM signal to the child process.
• kill(): kill the child process. Send SIGKILL signal to child process.

import time
import subprocess

def cmd(command):
    subp = subprocess.Popen(command,shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE,encoding="utf-8")
    subp.wait(2)
    if subp.poll() == 0:
        print(subp.communicate()[1])
    else:
        print("fail")

cmd("java -version")
cmd("exit 1")

The output results are as follows:

java version "1.8.0_31"
Java(TM) SE Runtime Environment (build 1.8.0_31-b13)
Java HotSpot(TM) 64-Bit Server VM (build 25.31-b07, mixed mode)

fail

Example

ping:

# Module to be imported: import subprocess [as alias]
# Or: from subprocess import run [as alias]
def ping(ns_name, source_lo_addr, dest_lo_addr):
    try:
        result = subprocess.run(['ip', 'netns', 'exec', ns_name, 'ping', '-f', '-W1',
                                 '-c{}'.format(PING_PACKTES),
                                 '-I', source_lo_addr, dest_lo_addr],
                                stdout=subprocess.PIPE)
    except FileNotFoundError:
        fatal_error('"ping" command not found')
    output = result.stdout.decode('ascii')
    lines = output.splitlines()
    for line in lines:
        if "packets transmitted" in line:
            split_line = line.split()
            packets_transmitted = int(split_line[0])
            packets_received = int(split_line[3])
            return (packets_transmitted, packets_received)
    fatal_error('Could not determine ping statistics for namespace "{}"'.format(ns_name))
    return None  # Never reached

popen:

# Module to be imported: import subprocess [as alias]
# Or: from subprocess import Popen [as alias]
def popen(cls, cmd, cwd=None, raises=False):
        '''
            Execute the given command string in a new process. Send data to stdin and
        read data from stdout and stderr, until end-of-file is reached.

        :param cls     : The class as implicit first argument.
        :param cwd     : If it is set, then the child's current directory will be change
                         to `cwd` before it is executed.
        :param raises  : If ``True`` and stderr has data, it raises an ``OSError`` exception.
        :returns       : The output of the given command; pair of (stdout, stderr).
        :rtype         : ``tuple``
        :raises OSError: If `raises` and stderr has data.
        '''
        parser   = lambda x: [] if x == '' else [y.strip() for y in x.strip().split('\n')]
        process  = subprocess.Popen(cmd, shell=True, universal_newlines=True,
                    stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        out, err = process.communicate()

        # .............................trim lines and remove the empty ones
        _stdout  = [x for x in parser(out) if bool(x)]
        _stderr  = [x for x in parser(err) if bool(x)]

        if _stderr and raises:
            raise OSError('\n'.join(_stderr))

        return _stdout, _stderr

runcmd:

import subprocess

'''
Runs a command, waits for it to complete, then returns a CompletedProcess instance.
'''

def runcmd(args,cwd=''):
    ret = subprocess.run(
        args=args,
        cwd=cwd,
        shell=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        encoding='gbk',
        timeout=1
    )
    if ret.returncode == 0:
        print("success:", ret.stdout)
    else:
        print("error:", ret.stderr)


if __name__ == "__main__":
    runcmd(args=["dir"],cwd=r'F:\test')

Result:
# success: the volume in drive F is data
#  The serial number of the volume is DCBE-C50C

#  F:\test directory

# 2021/12/27  19:01    <DIR>          .
# 2021/12/27  19:01    <DIR>          ..
# 2021/08/26  22:25            15,593 AutoComplete.cs
# 2021/08/26  22:25               706 ColorRandom.cs
# 2021/08/26  22:25             6,146 DataGridViewRender.cs
# 2021/08/26  22:25             2,840 ExcelDataReader.cs
# 2021/08/26  22:25            28,050 FTPClient.cs
# 2021/08/26  22:25               404 GUID.cs
# 2021/08/26  22:25               927 ListItem.cs
# 2021/08/26  22:25             1,360 LoadXml.cs
# 2021/08/26  22:25            35,700 ManageDB.cs
# 2021/08/26  22:25               428 RadomNamed.cs
# 2021/08/26  22:25             7,380 RegexInfo.cs
# 2021/08/26  22:25             4,111 SendMail.cs
# 2021/08/26  22:25             4,250 TextAndImageColumn.cs
# 2021/08/26  22:25             1,944 TimeDelay.cs
#               14 files 109839 bytes
#                2 directories 56677851136 bytes available