Python subprocess 模块
subprocess 是 Python 标准库中的一个模块,用于创建和管理子进程。
subprocess 允许你在 Python 程序中执行外部命令,并与这些命令进行交互。
通过 subprocess 模块,你可以执行系统命令、调用其他程序,并获取它们的输出或错误信息。
为什么使用 subprocess 模块?
在 Python 中,有时我们需要执行一些系统命令或调用其他程序来完成特定的任务。例如,你可能需要运行一个 shell 命令、启动一个外部应用程序,或者与一个命令行工具进行交互。subprocess 模块提供了一种安全且灵活的方式来处理这些需求。
与早期的 os.system() 或 os.popen() 相比,subprocess 模块提供了更强大的功能和更好的控制能力。它允许你更精细地管理子进程的输入、输出和错误流,并且可以处理更复杂的场景。
subprocess 模块的核心功能
1. 执行外部命令
subprocess.run() 是 subprocess 模块中最常用的函数之一。它可以执行一个外部命令,并等待命令完成。以下是一个简单的示例:
实例
# 执行一个简单的 shell 命令
result = subprocess.run(['ls', '-l'], capture_output=True, text=True)
# 打印命令的输出
print(result.stdout)
在这个例子中,subprocess.run() 执行了 ls -l 命令,并将输出捕获到 result.stdout 中。
2. 处理输入和输出
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 来获取子进程的退出状态码。
实例
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 来启动一个子进程,并在后台运行它,或者与它进行交互。
实例
# 启动一个子进程
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 模块允许你使用管道将多个命令连接在一起。你可以将一个命令的输出作为另一个命令的输入。
实例
# 使用管道连接两个命令
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.PIPE、open('log.txt', 'w') |
stderr | 标准错误配置 | subprocess.STDOUT(合并到stdout) |
shell | 是否通过Shell执行 | True(支持字符串命令) |
cwd | 工作目录路径 | "/tmp" |
env | 自定义环境变量 | {"PATH": "/usr/bin"} |
timeout | 超时时间(秒) | 30 |
text | 输入/输出是否为字符串(非字节) | True |
实例
执行命令并捕获输出:
实例
print(result.stdout) # 输出: "Hello\n"
通过 Shell 执行复杂命令:
subprocess.run("grep 'error' log.txt | wc -l", shell=True, check=True)
实时获取输出流:
实例
while True:
line = proc.stdout.readline()
if not line: break
print(line.decode().strip())
超时控制:
实例
subprocess.run(["sleep", "10"], timeout=5)
except subprocess.TimeoutExpired:
print("命令超时!")