subprocess
info
在任何编程语言中,启动新进程都是一项常见任务,在像 Python 这样的高级语言中更是如此。因此,我们需要对这项任务提供良好的支持,原因如下:
- 启动进程时使用不合适的函数可能意味着安全风险:如果程序通过 shell 启动,并且参数包含 shell 元字符,则后果可能不堪设想。
- 这使得 Python 成为替代过于复杂的 shell 脚本的更佳语言。
run
最推荐的用法,简单直接。
import subprocess
# 执行命令
result = subprocess.run(['ls', '-l'], capture_output=True, text=True)
print(result.stdout) # 标准输出
print(result.stderr) # 错误输出
print(result.returncode) # 返回码,0表示成功
常用参数:
args: 要执行的命令,列表或字符串shell: 是否通过 shell 执行,默认 Falsecapture_output: 捕获输出,等同于stdout=PIPE, stderr=PIPEtext: 以文本模式返回,不用再 decodetimeout: 超时时间(秒),超时会抛异常check: True 时,如果命令返回非 0 会抛异常
有时候需要执行管道、重定向等 shell 特性:
import subprocess
# 这种情况需要 shell=True
result = subprocess.run('ls -l | grep py', shell=True, capture_output=True, text=True)
print(result.stdout)
# 或者传字符串
result = subprocess.run('echo "Hello World"', shell=True, capture_output=True, text=True)
print(result.stdout)
warning
shell=True 有安全风险,不要传入用户输入的内容,避免命令注入。
处理错误
import subprocess
try:
result = subprocess.run(
['python', 'nonexist.py'],
capture_output=True,
text=True,
check=True, # 返回码非0时抛异常
timeout=5
)
except subprocess.CalledProcessError as e:
print(f"命令执行失败,返回码: {e.returncode}")
print(f"错误输出: {e.stderr}")
except subprocess.TimeoutExpired:
print("命令执行超时")
Popen() 方法
更底层的接口,用于需要精细控制的场景。
import subprocess
# 启动进程
process = subprocess.Popen(
['python', '-u', 'script.py'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# 实时读取输出
for line in process.stdout:
print(line, end='')
# 等待进程结束
return_code = process.wait()
print(f"进程结束,返回码: {return_code}")
与进程交互
import subprocess
process = subprocess.Popen(
['python', '-u'], # -u 让输出不缓冲
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# 发送输入并获取输出
stdout, stderr = process.communicate(input='print("Hello from stdin")\n')
print(stdout)
实例
将脚本输出重定向
子进程里所有 print() 默认写到标准输出(stdout)。
用 subprocess.run 时把 stdout 设为一个已打开的文件对象,即可把所有 print() 输出写入该文件。
import subprocess
# 把所有 print 输出写入 output.txt(仅 stdout)
with open('output.txt', 'w', encoding='utf-8') as f:
subprocess.run(
['python', 'script.py'],
stdout=f,
text=True
)
# 若还想把 stderr 也写入同一文件:
# stdout=f, stderr=f
若希望 stdout 和 stderr 都写入同一文件:
import subprocess
with open('output.txt', 'w', encoding='utf-8') as f:
subprocess.run(
['python', 'script.py'],
stdout=f,
stderr=subprocess.STDOUT, # stderr 也写入 f
text=True
)
要点:
stdout=open('output.txt', 'w', ...)表示子进程的 stdout 接到该文件,等价于在命令行里python script.py > output.txt。- 使用
text=True时,用encoding='utf-8'打开文件,避免编码问题。 - 用
with open(...)保证子进程结束后文件被正确关闭。
脚本报错时:
- 当前进程不会抛异常:默认没有
check=True时,子进程就算非零退出(如未捕获异常、sys.exit(1)),subprocess.run()也只会正常返回,不会在你这边“弹错”。需要自己看result.returncode或设check=True才会在失败时抛CalledProcessError。 - 错误输出是否“弹出”:Python 的 traceback 和很多错误信息是写到 stderr 的。若只重定向了
stdout,stderr 仍会打印到当前终端;要想错误也进文件、终端不弹出,需要同时重定向 stderr(例如stderr=f或stderr=subprocess.STDOUT)。
执行 git 命令
import subprocess
def git_status():
result = subprocess.run(
# 简短的查看git提交变动
['git', 'status', '--short'],
capture_output=True,
text=True
)
return result.stdout
def git_commit(message):
subprocess.run(['git', 'add', '.'], check=True)
subprocess.run(['git', 'commit', '-m', message], check=True)
print(git_status())
调用其他 Python 脚本
一个文件处理脚本
data_process.py
import argparse
# 1. 创建解析器
parser = argparse.ArgumentParser()
# 2. 定义程序接受哪些参数
parser.add_argument('--input', help='输入文件的路径')
# 3. 解析参数
args = parser.parse_args()
# 4. 使用参数
print(f"正在处理文件: {args.input}")
# 如果你传了 data.csv,这里的 args.input 就是 "data.csv"
通过subprocess调用这个脚本
import subprocess
result = subprocess.run(
#
['python', 'data_process.py', '--input', 'data.csv'],
capture_output=True,
text=True,
timeout=300 # 5分钟超时
)
if result.returncode == 0:
print("处理成功")
print(result.stdout)
else:
print("处理失败")
print(result.stderr)