Python 标准模块 - subprocess

2023-10-18

字数统计: 1.7k字 | 阅读时长≈ 7分

subprocess 是 Python 自带的模块，无须安装，它主要用来取代一些旧的模块或方法，如 os.system, os.spawn*, os.popen*, commands.*等，因此如果需要使用 Python 调用外部命令或任务时，则优先使用 subprocess 模块。使用 subprocess 模块可以方便地执行操作系统支持的命令，可与其他应用程序结合使用。

subprocess.run() 方法

subprocess.run() 是官方推荐使用的方法，几乎所有的工作都可以由它来完成。

该函数的原型如下:

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, 
    **other_popen_kwargs)

虽然该函数的参数有很多，但是我们只需要知道几个常用的就可以了，如下:

args: 代表需要在操作系统中执行的命令，可以是字符串形式（要求 shell=True），也可以是列表 list 类型;
*: 代表可变参数，一般是列表或字典形式;
stdin, stdout, stderr: 指定看可执行程序的标准输入，标准输出，标准错误文件句柄;
shell: 代表着程序是否需要再 shell 上执行，当想使用 shell 的特性时，设置 shell=True，这样可以使用 shell 指令的管道，文件名称通配符，环境变量等，不过 Python 也提供了许多类 shell 的模块，如 glob、fnmatch、os.walk()、os.path.expandvars()、os.path.expanduser() 和 shutil。
check: 如果 check 设置为 True，就检查命令的返回值，当返回值为非 0 时，将抛出 CalledProcessError 异常;
timeout: 设置超时时间，如果超时，则强制 kill 掉子进程。

CompletedProcess 类

run() 方法的返回值，表示一个进程结束了。CompletedProcess 类有下面这些属性：

args: 启动进程的参数，通常是个列表或字符串
returncode: 进程结束状态返回码。0 表示成功状态
stdout: 获取子进程的stdout。通常为 bytes 类型序列，None 表示没有捕获值。如果在调用 run() 方法时，设置了参数 stderr=subprocess.STDOUT，则错误信息会和 stdout 一起输出，此时 stderr 的值是 None
stderr: 获取子进程的错误信息，通常为 bytes 类型序列，None 表示没有捕获值
check_returncode(): 用于检查返回码。如果返回状态码不为零，弹出 CalledProcessError 异常

示例

在 Linux 系统中，如果我们执行一个脚本并获取它的返回值，可以有以下两种方法，如下

In [1]: import subprocess

# 方法-: 使用字符串形式传递命令
In [1]: a = subprocess.run("ls -l /dev/null", shell=True)
crw-rw-rw-  1 root  wheel  0x3000002  3 22 10:16 /dev/null

In [3]: a
Out[3]: CompletedProcess(args='ls -l /dev/null', returncode=0)

In [4]: a.args
Out[4]: 'ls -l /dev/null'

In [5]: a.returncode
Out[5]: 0

# 方法二: 使用列表形式传递命令
In [7]: b = subprocess.run(["ls", "-l", "/dev/null"])
crw-rw-rw-  1 root  wheel  0x3000002  3 22 10:17 /dev/null

In [8]: b
Out[8]: CompletedProcess(args=['ls', '-l', '/dev/null'], returncode=0)

In [9]: b.args
Out[9]: ['ls', '-l', '/dev/null']

In [10]: b.returncode
Out[10]: 0

如果要获取脚本的输出，可以使用如下方法

In [14]: a = subprocess.run("ls -l /dev/null", shell=True, stdout=subprocess.PIPE)

In [15]: a
Out[15]: CompletedProcess(args='ls -l /dev/null', returncode=0, stdout=b'crw-rw-rw-  1 root  wheel  0x3000002  3 22 10:36 /dev/null\n')

In [16]: a.stdout
Out[16]: b'crw-rw-rw-  1 root  wheel  0x3000002  3 22 10:36 /dev/null\n'

如果传入参数 checkout=True，当 returncode 不为 0 时，将会抛出 subprocess.CalledProcessError 异常；如果传入 timeout 参数，当运行时间超过 timeout 时，就会抛出 TimeoutExpired 异常。如下

# CalledProcessError 异常
In [24]: a = subprocess.run(["ifconfig", "a"], check=True)
ifconfig: interface a does not exist
---------------------------------------------------------------------------
CalledProcessError                        Traceback (most recent call last)
Cell In[24], line 1
----> 1 a = subprocess.run(["ifconfig", "a"], check=True)

File ~/.pyenv/versions/3.10.10/lib/python3.10/subprocess.py:526, in run(input, capture_output, timeout, check, *popenargs, **kwargs)
    524     retcode = process.poll()
    525     if check and retcode:
--> 526         raise CalledProcessError(retcode, process.args,
    527                                  output=stdout, stderr=stderr)
    528 return CompletedProcess(process.args, retcode, stdout, stderr)

CalledProcessError: Command '['ifconfig', 'a']' returned non-zero exit status 1.


# TimeoutExpired 异常
In [25]: a = subprocess.run(["sleep", "5"], timeout=2)
---------------------------------------------------------------------------
TimeoutExpired                            Traceback (most recent call last)
Cell In[25], line 1
----> 1 a = subprocess.run(["sleep", "5"], timeout=2)

... 省略 N 行 ...
File ~/.pyenv/versions/3.10.10/lib/python3.10/subprocess.py:1935, in Popen._wait(self, timeout)
  1933 remaining = self._remaining_time(endtime)
  1934 if remaining <= 0:
-> 1935     raise TimeoutExpired(self.args, timeout)
  1936 delay = min(delay * 2, remaining, .05)
  1937 time.sleep(delay)

TimeoutExpired: Command '['sleep', '5']' timed out after 1.9996612090035342 seconds

subprocess.Popen() 类

Popen 类的构造函数如下

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=None, 
    startupinfo=None, 
    creationflags=0, 
    restore_signals=True, 
    start_new_session=False, 
    pass_fds=(), 
    *, 
    group=None, 
    extra_groups=None, 
    user=None, 
    umask=- 1, 
    encoding=None, 
    errors=None, 
    text=None, 
    pipesize=- 1, 
    process_group=None)

常用的参数如下:

bufsize: 0 为无缓冲，1 为行缓冲，其他正值表示缓冲区大小，负值表示采用默认系统缓冲（一般是全缓冲）;
executable: 一般不用;
args: 字符串或列表第一项表示程序名;
stdin, stdout, stderr: None 表示没有任何重定向，继承父进程；PIPE 表示创建管道；文件对象；文件描述符（整数）；stderr 还可以设置为 STDOUT;
preexec_fn: 钩子函数，在 fork 和 exec 之间执行;
close_fds: unix 下执行新进程前是否关闭 0/1/2 之外的文件；windows 下不继承还是继承父进程的文件描述符;
shell: 为真的话，unix 下相当于 args 前面添加了 “/bin/sh” “-c”，Windows 下相当于添加了 “cmd.exe /c”;
cwd: 设置工作目录;
env: 设置环境变量;
universal_newlines: 各种换行符统一处理成 ‘\n’;
startupinfo: Windows 下传递给 CreateProcess 的结构体;
creationflags: Windows 下传递 CREATE_NEW_CONSOLE 创建自己的控制台窗口;

Popen 使用方法如下

1 2	In [34]: subprocess.Popen(["cat", "wb.txt"]) In [36]: subprocess.Popen("cat wb.txt", shell=True)

Popen 类的对象还有其他实用方法，如下

名称	功能
poll()	检查是否结束，设置返回值
wait()	等待结束，设置返回值
communicate()	参数是标准输入，返回标准输出和标准出错
send_signal()	发送信号（主要是在 unix 下使用）
terminate()	终止进程，unix 对应的 SIGTERM 信号，Windows 下调用 api 函数 TerminateProcess()
kill()	杀死进程(unix 对应 SIGKILL 信号)。Windows 下同上
stdin, stdout, stderr	参数中指定 PIPE 时，可以使用
pid	进程的 id
returncode	进程返回值

subprocess 的其他方法

subprocess.call(): 调用 Popen() 执行程序，并等待它完成;
subprocess.check_call(): 调用前面的 call()，如果返回值为非零，则抛出异常;
subprocess.check_output(): 调用 Popen() 执行程序，并返回其标准输出