当前位置：首页 > news >正文

Python subprocess 模块详解

news 2025/9/11 14:07:02

Python subprocess 模块详解

subprocess 模块是 Python 中用于创建和管理子进程的标准库，它允许你启动新的应用程序、连接到它们的输入/输出/错误管道，并获取它们的返回码。

基本用法

1. 运行简单命令

import subprocess# 运行命令并等待完成
result = subprocess.run(['ls', '-l'])
print(result)

2. 捕获输出

result = subprocess.run(['ls', '-l'], capture_output=True, text=True)
print(result.stdout)

3. 检查返回码

result = subprocess.run(['ls', '/nonexistent'], capture_output=True, text=True)
if result.returncode != 0:print(f"命令失败，错误信息: {result.stderr}")

主要函数

subprocess.run()

最常用的函数，运行命令并等待完成：

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

参数说明：

args: 命令列表或字符串（当shell=True时）
capture_output: 捕获stdout和stderr
text: 以文本模式处理输入输出（Python 3.7+）
check: 如果为True，当返回码非零时抛出异常

其他常用函数

subprocess.call(): 运行命令并等待完成，返回返回码
subprocess.check_call(): 类似call()，但当返回码非零时抛出异常
subprocess.check_output(): 运行命令并返回输出
subprocess.Popen(): 更底层的控制，用于复杂场景

高级用法

1. 管道通信

p1 = subprocess.Popen(['ls', '-l'], stdout=subprocess.PIPE)
p2 = subprocess.Popen(['grep', '.py'], stdin=p1.stdout, stdout=subprocess.PIPE)
p1.stdout.close()  # 允许p1接收SIGPIPE信号
output = p2.communicate()[0]
print(output.decode('utf-8'))

2. 交互式进程

process = subprocess.Popen(['python3'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,text=True)process.stdin.write("print('Hello from subprocess!')\n")
process.stdin.write("exit()\n")
output = process.communicate()[0]
print(output)

3. 处理超时

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

安全注意事项

避免使用shell=True：当使用shell=True时，命令会通过shell执行，可能带来shell注入风险

# 不安全
subprocess.run(f'echo {user_input}', shell=True)# 更安全的方式
subprocess.run(['echo', user_input])

如果必须使用shell特性，考虑使用shlex.quote()对参数进行转义

实际示例

检查网络连接

import subprocess
import shlexdef check_ping(host):command = f"ping -c 1 {host}"try:subprocess.run(command, shell=True, check=True, timeout=5)print(f"{host} 可达")except (subprocess.CalledProcessError, subprocess.TimeoutExpired):print(f"{host} 不可达或超时")

下载文件

import subprocessdef download_file(url, output_path):try:subprocess.run(['curl', '-o', output_path, url], check=True)print("下载成功")except subprocess.CalledProcessError as e:print(f"下载失败: {e}")