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 执行，默认 False
capture_output: 捕获输出，等同于 stdout=PIPE, stderr=PIPE
text: 以文本模式返回，不用再 decode
timeout: 超时时间（秒），超时会抛异常
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)

实际例子

执行 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)

run​

处理错误​

Popen() 方法​

与进程交互​

实际例子​

执行 git 命令​

调用其他 Python 脚本​

run