Python subprocess 模块

subprocess 是 Python 标准库中的一个模块,用于创建和管理子进程。

subprocess 允许你在 Python 程序中执行外部命令,并与这些命令进行交互。

通过 subprocess 模块,你可以执行系统命令、调用其他程序,并获取它们的输出或错误信息。

为什么使用 subprocess 模块?

在 Python 中,有时我们需要执行一些系统命令或调用其他程序来完成特定的任务。例如,你可能需要运行一个 shell 命令、启动一个外部应用程序,或者与一个命令行工具进行交互。subprocess 模块提供了一种安全且灵活的方式来处理这些需求。

与早期的 os.system()os.popen() 相比,subprocess 模块提供了更强大的功能和更好的控制能力。它允许你更精细地管理子进程的输入、输出和错误流,并且可以处理更复杂的场景。


subprocess 模块的核心功能

1. 执行外部命令

subprocess.run()subprocess 模块中最常用的函数之一。它可以执行一个外部命令,并等待命令完成。以下是一个简单的示例:

实例

import subprocess

# 执行一个简单的 shell 命令
result = subprocess.run(['ls', '-l'], capture_output=True, text=True)

# 打印命令的输出
print(result.stdout)

在这个例子中,subprocess.run() 执行了 ls -l 命令,并将输出捕获到 result.stdout 中。

2. 处理输入和输出

subprocess 模块允许你控制子进程的输入、输出和错误流。你可以将数据传递给子进程的标准输入,或者从子进程的标准输出和标准错误中读取数据。以下是一个示例:

实例

import subprocess

# 执行一个命令,并将输入传递给子进程
result = subprocess.run(['grep', 'python'], input='hello\npython\nworld', capture_output=True, text=True)

# 打印命令的输出
print(result.stdout)

在这个例子中,subprocess.run() 执行了 grep python 命令,并将字符串 'hello\npython\nworld' 作为输入传递给子进程。

3. 处理错误

subprocess 模块还允许你处理子进程的错误。如果子进程返回非零的退出状态码,subprocess.run() 会抛出一个 CalledProcessError 异常。你可以通过检查 result.returncode 来获取子进程的退出状态码。

实例

import subprocess

try:
    result = subprocess.run(['ls', 'nonexistent_file'], capture_output=True, text=True, check=True)
except subprocess.CalledProcessError as e:
    print(f"Command failed with return code {e.returncode}")
    print(f"Error output: {e.stderr}")

在这个例子中,subprocess.run() 执行了 ls nonexistent_file 命令,由于文件不存在,命令失败并抛出了 CalledProcessError 异常。


subprocess 模块的高级用法

1. 使用 Popen 类

subprocess.Popen 类提供了更底层的接口,允许你更灵活地控制子进程。你可以使用 Popen 来启动一个子进程,并在后台运行它,或者与它进行交互。

实例

import subprocess

# 启动一个子进程
process = subprocess.Popen(['ping', 'google.com'], stdout=subprocess.PIPE, text=True)

# 读取子进程的输出
while True:
    output = process.stdout.readline()
    if output == '' and process.poll() is not None:
        break
    if output:
        print(output.strip())

# 获取子进程的退出状态码
return_code = process.poll()
print(f"Process finished with return code {return_code}")

在这个例子中,subprocess.Popen 启动了一个 ping google.com 命令,并在后台运行它。程序通过循环读取子进程的输出,并在子进程结束后获取其退出状态码。

2. 使用管道

subprocess 模块允许你使用管道将多个命令连接在一起。你可以将一个命令的输出作为另一个命令的输入。

实例

import subprocess

# 使用管道连接两个命令
p1 = subprocess.Popen(['ls', '-l'], stdout=subprocess.PIPE)
p2 = subprocess.Popen(['grep', 'py'], stdin=p1.stdout, stdout=subprocess.PIPE, text=True)

# 获取最终输出
output = p2.communicate()[0]
print(output)

在这个例子中,ls -l 命令的输出被传递给 grep py 命令,最终输出包含 py 的文件或目录。


subprocess 模块的常用方法、类和参数

以下是 Python subprocess 模块的常用方法、类和参数的说明,包含功能描述及示例:

subprocess 模块核心方法

方法说明示例
subprocess.run()执行命令并等待完成(推荐)subprocess.run(["ls", "-l"], capture_output=True, text=True)
subprocess.Popen()创建子进程(底层控制)proc = subprocess.Popen(["ping", "google.com"], stdout=subprocess.PIPE)
subprocess.call()执行命令并返回退出码(旧版)exit_code = subprocess.call(["python", "--version"])
subprocess.check_call()执行命令,失败时抛出异常subprocess.check_call(["git", "commit"])
subprocess.check_output()执行命令并返回输出(旧版)output = subprocess.check_output(["date"], text=True)

subprocess.CompletedProcess 对象属性(run() 方法的返回对象)

属性说明
args执行的命令参数列表
returncode进程退出状态码(0表示成功)
stdout标准输出内容(若设置了capture_output
stderr标准错误内容(若设置了capture_output

subprocess.Popen 类常用方法/属性

方法/属性说明示例
poll()检查进程是否终止(返回None表示运行中)if proc.poll() is None: print("Running")
wait()阻塞等待进程结束proc.wait()
communicate()交互式输入/输出stdout, stderr = proc.communicate(input="data")
terminate()发送终止信号(SIGTERM)proc.terminate()
kill()强制终止进程(SIGKILL)proc.kill()
stdin进程的标准输入流proc.stdin.write("input")
stdout进程的标准输出流print(proc.stdout.read())
stderr进程的标准错误流errors = proc.stderr.read()

常用参数说明(适用于 run() 和 Popen())

参数说明示例值
args命令(列表或字符串)["ls", "-l"]"ls -l"
stdin标准输入配置subprocess.PIPE(管道)、None(继承)
stdout标准输出配置subprocess.PIPEopen('log.txt', 'w')
stderr标准错误配置subprocess.STDOUT(合并到stdout)
shell是否通过Shell执行True(支持字符串命令)
cwd工作目录路径"/tmp"
env自定义环境变量{"PATH": "/usr/bin"}
timeout超时时间(秒)30
text输入/输出是否为字符串(非字节)True

实例

执行命令并捕获输出:

实例

result = subprocess.run(["echo", "Hello"], capture_output=True, text=True)
print(result.stdout)  # 输出: "Hello\n"

通过 Shell 执行复杂命令:

subprocess.run("grep 'error' log.txt | wc -l", shell=True, check=True)
实时获取输出流:

实例

proc = subprocess.Popen(["tail", "-f", "log.txt"], stdout=subprocess.PIPE)
while True:
    line = proc.stdout.readline()
    if not line: break
    print(line.decode().strip())

超时控制:

实例

try:
    subprocess.run(["sleep", "10"], timeout=5)
except subprocess.TimeoutExpired:
    print("命令超时!")