subprocess模块是 Python 官方推荐用于创建和管理子进程的标准库它统一并替代了os.system、os.popen等旧式方法。它能让你安全、灵活地执行外部命令并获取其输入输出。 核心函数run()与Popen()subprocess提供了两个主要接口你可以按需选用subprocess.run()(推荐): 这是 Python 3.5 之后的首选方式。它用于执行一个命令并等待其完成返回一个包含执行结果的CompletedProcess对象。适合大多数一次性调用场景。subprocess.Popen()(高级): 这是更底层的接口run()也是基于它实现的-。Popen对象代表一个正在运行的子进程允许你在进程启动后与之持续交互、检查状态等。适合需要复杂交互、实时流式处理或非阻塞式管理的场景。run()常用参数详解:subprocess.run(args,*,stdinNone,inputNone,stdoutNone,stderrNone,capture_outputFalse,shellFalse,cwdNone,timeoutNone,checkFalse,encodingNone,errorsNone,textNone,envNone,universal_newlinesNone,**other_popen_kwargs)运行被args描述的指令。等待指令完成然后返回一个 CompletedProcess 实例。参数说明示例args要执行的命令推荐使用列表形式如[ls, -l]更安全。subprocess.run([echo, Hello])capture_output若为True则捕获子进程的stdout和stderr。subprocess.run(..., capture_outputTrue)text若为True则输入输出以文本 (字符串)形式处理否则为bytes字节流。subprocess.run(..., textTrue)check若为True当子进程返回非零退出码时抛出CalledProcessError异常。subprocess.run(..., checkTrue)timeout设置超时时间秒超时后会抛出TimeoutExpired异常。subprocess.run(..., timeout10)input向子进程的标准输入发送数据。subprocess.run(..., inputy\n)env为子进程设置自定义的环境变量字典。subprocess.run(..., env{PATH: /custom/bin})cwd设置子进程的当前工作目录。subprocess.run(..., cwd/path/to/dir) 安全第一警惕shellTrue这是使用subprocess时最需要警惕的风险点。风险: 当shellTrue时命令会通过系统的 shell如/bin/sh或cmd.exe执行。这意味着命令字符串中的特殊字符如;,|,,$等会被 shell 解析极易引发命令注入攻击。错误示例:# 极度危险切勿使用 user_input file.txt; rm -rf / subprocess.run(fcat {user_input}, shellTrue) # 可能执行删除操作正确做法:尽量避免使用shellTrue。大多数情况下直接传递参数列表是安全且有效的。# 安全做法 filename file with space.txt subprocess.run([cat, filename]) # 参数列表安全可靠何时使用: 仅当命令必须依赖 shell 特性如通配符*.txt、管道|、环境变量$HOME且命令本身是静态的、不包含任何用户输入时才考虑使用。️ 核心技巧与进阶用法1. 捕获与处理输出分别捕获: 使用capture_outputTrue可以分别获取stdout和stderr。合并输出: 如果你想将标准错误合并到标准输出中可以设置stderrsubprocess.STDOUT。实时输出: 对于耗时命令run()会等待进程结束才返回。如需实时获取输出需使用Popen逐行读取stdout。import subprocess # 启动进程 process subprocess.Popen( [ping, -c, 5, google.com], stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, bufsize1 # 行缓冲 ) # 实时读取输出逐行 while True: output process.stdout.readline() if output and process.poll() is not None: break if output: print(output.strip()) # 获取剩余输出和返回码 stdout, stderr process.communicate()2. 输入与管道提供输入: 通过input参数可以向子进程的标准输入发送数据。subprocess.run([grep, error], inputline1\nerror line, textTrue)构建管道: 如需实现 Shell 中的管道如cmd1 | cmd2可以手动创建两个Popen对象并将第一个的stdout连接到第二个的stdin。使用subprocess实现 Shell 管道cmd1 | cmd2的核心原理是将第一个进程的stdout连接到第二个进程的stdin。在 Python 中最标准且安全的方法是手动创建两个Popen对象并利用subprocess.PIPE来连接它们。下面我将从最基础的实现开始逐步深入到链式管道和错误处理。核心实现连接两个命令下面以echo hello world | grep world为例import subprocess # 1. 启动第一个进程将其 stdout 重定向到管道 (PIPE) p1 subprocess.Popen( [echo, hello world], stdoutsubprocess.PIPE, textTrue # 以文本模式处理 ) # 2. 启动第二个进程将其 stdin 设置为 p1 的 stdout p2 subprocess.Popen( [grep, world], stdinp1.stdout, # 关键连接点从 p1 读取输出 stdoutsubprocess.PIPE, textTrue ) # ⚠️ 极其关键的一步在父进程中关闭 p1 的 stdout p1.stdout.close() # 3. 获取最终结果从 p2 读取 output, error p2.communicate() print(f输出: {output}) # 输出: hello world print(fp1 返回码: {p1.wait()}) # 通常为 0 print(fp2 返回码: {p2.returncode}) # 0为什么必须关闭p1.stdout.close()p1.stdout在父进程你的 Python 脚本中也持有管道的读取端。如果不关闭它当p2处理完数据退出后p1可能仍在尝试写入数据。更严重的是如果p2提前退出比如只匹配前几行就退出p1会因为管道缓冲区写满而永久挂起因为管道的另一端父进程持有的那个始终没有关闭。关闭后当p2退出时p1会收到SIGPIPE信号并自动终止或者正常写完所有数据后退出。进阶实现链式管道cmd1 | cmd2 | cmd3假设我们要实现cat file.txt | grep error | wc -l只需按顺序串联即可。核心原则是始终关闭上一个进程的stdout。import subprocess # 构建管道链 p1 subprocess.Popen([cat, file.txt], stdoutsubprocess.PIPE, textTrue) p2 subprocess.Popen([grep, error], stdinp1.stdout, stdoutsubprocess.PIPE, textTrue) p1.stdout.close() # p1 的输出已交给 p2父进程不再需要 p3 subprocess.Popen([wc, -l], stdinp2.stdout, stdoutsubprocess.PIPE, textTrue) p2.stdout.close() # p2 的输出已交给 p3父进程不再需要 # 获取最终输出 result, _ p3.communicate() print(f错误行数: {result.strip()}) # 确保所有进程都被回收如果它们还没退出 p1.wait() p2.wait() p3.wait()封装为通用函数便于复用import subprocess def safe_pipeline(commands): procs [] for i, cmd in enumerate(commands): if i 0: p subprocess.Popen(cmd, stdoutsubprocess.PIPE, textTrue) else: p subprocess.Popen(cmd, stdinprocs[-1].stdout, stdoutsubprocess.PIPE, textTrue) procs[-1].stdout.close() procs.append(p) output, _ procs[-1].communicate() # 等待所有进程结束并检查返回码 for p in procs: p.wait() if p.returncode ! 0: # 抛出异常或进行错误处理注意管道中 grep 无匹配时返回 1 是正常的 raise subprocess.CalledProcessError(p.returncode, p.args) return output try: # 故意让 grep 匹配不到会返回 1 result safe_pipeline([ [echo, abc], [grep, xyz] ]) except subprocess.CalledProcessError as e: print(f管道中的命令失败: {e})补充技巧同时获取 stderr如果你想分别获取每个命令的错误信息可以为每个Popen设置stderrsubprocess.PIPE然后通过p.communicate()或后续读取p.stderr来获取。3. 超时控制为run()或Popen.communicate()设置timeout参数可以防止子进程永久挂起-。try: subprocess.run([sleep, 10], timeout5) except subprocess.TimeoutExpired: print(命令执行超时)4. 环境变量与工作目录环境变量: 使用env参数可以为子进程设置独立的变量而不是继承父进程的。工作目录: 使用cwd参数可以指定子进程的执行路径。5. 错误处理设置checkTrue可以让run()在命令失败时抛出异常这比手动检查returncode更简洁。try: subprocess.run([false], checkTrue) except subprocess.CalledProcessError as e: print(f命令失败返回码: {e.returncode})⚠️ 避坑指南避免字符串拼接: 构建命令参数时始终使用列表而非字符串拼接以防止注入和参数解析错误。注意文本与字节: 使用textTrue处理文本使用bytes处理二进制数据。跨平台兼容性: Windows 和 Unix 的命令、路径风格不同编写脚本时需注意。例如Windows 上可执行文件通常有.exe后缀。僵尸进程: 使用Popen时务必调用proc.wait()或proc.communicate()来回收子进程资源避免产生僵尸进程-。PIPE死锁: 如果使用PIPE且数据量很大必须及时读取否则可能导致死锁。communicate()方法可以安全地处理这个问题-。import subprocess # 启动进程 process subprocess.Popen( [ping, -c, 5, google.com], stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, bufsize1 # 行缓冲 ) # 实时读取输出逐行 while True: output process.stdout.readline() if output and process.poll() is not None: break if output: print(output.strip()) # 获取剩余输出和返回码 stdout, stderr process.communicate()Popen作为上下文管理器: 从 Python 3.2 开始Popen支持上下文管理器可以确保资源被正确清理-。with subprocess.Popen([cmd], stdoutsubprocess.PIPE) as proc: print(proc.communicate()[0]) 总结使用subprocess模块记住以下三点优先使用subprocess.run()它足以应对 90% 的需求。默认使用参数列表除非绝对必要否则永远不要开启shellTrue。善用capture_output、text和check等参数编写更健壮的代码。参考资料https://docs.python.org/zh-cn/3/library/subprocess.html
Python subprocess详解
subprocess模块是 Python 官方推荐用于创建和管理子进程的标准库它统一并替代了os.system、os.popen等旧式方法。它能让你安全、灵活地执行外部命令并获取其输入输出。 核心函数run()与Popen()subprocess提供了两个主要接口你可以按需选用subprocess.run()(推荐): 这是 Python 3.5 之后的首选方式。它用于执行一个命令并等待其完成返回一个包含执行结果的CompletedProcess对象。适合大多数一次性调用场景。subprocess.Popen()(高级): 这是更底层的接口run()也是基于它实现的-。Popen对象代表一个正在运行的子进程允许你在进程启动后与之持续交互、检查状态等。适合需要复杂交互、实时流式处理或非阻塞式管理的场景。run()常用参数详解:subprocess.run(args,*,stdinNone,inputNone,stdoutNone,stderrNone,capture_outputFalse,shellFalse,cwdNone,timeoutNone,checkFalse,encodingNone,errorsNone,textNone,envNone,universal_newlinesNone,**other_popen_kwargs)运行被args描述的指令。等待指令完成然后返回一个 CompletedProcess 实例。参数说明示例args要执行的命令推荐使用列表形式如[ls, -l]更安全。subprocess.run([echo, Hello])capture_output若为True则捕获子进程的stdout和stderr。subprocess.run(..., capture_outputTrue)text若为True则输入输出以文本 (字符串)形式处理否则为bytes字节流。subprocess.run(..., textTrue)check若为True当子进程返回非零退出码时抛出CalledProcessError异常。subprocess.run(..., checkTrue)timeout设置超时时间秒超时后会抛出TimeoutExpired异常。subprocess.run(..., timeout10)input向子进程的标准输入发送数据。subprocess.run(..., inputy\n)env为子进程设置自定义的环境变量字典。subprocess.run(..., env{PATH: /custom/bin})cwd设置子进程的当前工作目录。subprocess.run(..., cwd/path/to/dir) 安全第一警惕shellTrue这是使用subprocess时最需要警惕的风险点。风险: 当shellTrue时命令会通过系统的 shell如/bin/sh或cmd.exe执行。这意味着命令字符串中的特殊字符如;,|,,$等会被 shell 解析极易引发命令注入攻击。错误示例:# 极度危险切勿使用 user_input file.txt; rm -rf / subprocess.run(fcat {user_input}, shellTrue) # 可能执行删除操作正确做法:尽量避免使用shellTrue。大多数情况下直接传递参数列表是安全且有效的。# 安全做法 filename file with space.txt subprocess.run([cat, filename]) # 参数列表安全可靠何时使用: 仅当命令必须依赖 shell 特性如通配符*.txt、管道|、环境变量$HOME且命令本身是静态的、不包含任何用户输入时才考虑使用。️ 核心技巧与进阶用法1. 捕获与处理输出分别捕获: 使用capture_outputTrue可以分别获取stdout和stderr。合并输出: 如果你想将标准错误合并到标准输出中可以设置stderrsubprocess.STDOUT。实时输出: 对于耗时命令run()会等待进程结束才返回。如需实时获取输出需使用Popen逐行读取stdout。import subprocess # 启动进程 process subprocess.Popen( [ping, -c, 5, google.com], stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, bufsize1 # 行缓冲 ) # 实时读取输出逐行 while True: output process.stdout.readline() if output and process.poll() is not None: break if output: print(output.strip()) # 获取剩余输出和返回码 stdout, stderr process.communicate()2. 输入与管道提供输入: 通过input参数可以向子进程的标准输入发送数据。subprocess.run([grep, error], inputline1\nerror line, textTrue)构建管道: 如需实现 Shell 中的管道如cmd1 | cmd2可以手动创建两个Popen对象并将第一个的stdout连接到第二个的stdin。使用subprocess实现 Shell 管道cmd1 | cmd2的核心原理是将第一个进程的stdout连接到第二个进程的stdin。在 Python 中最标准且安全的方法是手动创建两个Popen对象并利用subprocess.PIPE来连接它们。下面我将从最基础的实现开始逐步深入到链式管道和错误处理。核心实现连接两个命令下面以echo hello world | grep world为例import subprocess # 1. 启动第一个进程将其 stdout 重定向到管道 (PIPE) p1 subprocess.Popen( [echo, hello world], stdoutsubprocess.PIPE, textTrue # 以文本模式处理 ) # 2. 启动第二个进程将其 stdin 设置为 p1 的 stdout p2 subprocess.Popen( [grep, world], stdinp1.stdout, # 关键连接点从 p1 读取输出 stdoutsubprocess.PIPE, textTrue ) # ⚠️ 极其关键的一步在父进程中关闭 p1 的 stdout p1.stdout.close() # 3. 获取最终结果从 p2 读取 output, error p2.communicate() print(f输出: {output}) # 输出: hello world print(fp1 返回码: {p1.wait()}) # 通常为 0 print(fp2 返回码: {p2.returncode}) # 0为什么必须关闭p1.stdout.close()p1.stdout在父进程你的 Python 脚本中也持有管道的读取端。如果不关闭它当p2处理完数据退出后p1可能仍在尝试写入数据。更严重的是如果p2提前退出比如只匹配前几行就退出p1会因为管道缓冲区写满而永久挂起因为管道的另一端父进程持有的那个始终没有关闭。关闭后当p2退出时p1会收到SIGPIPE信号并自动终止或者正常写完所有数据后退出。进阶实现链式管道cmd1 | cmd2 | cmd3假设我们要实现cat file.txt | grep error | wc -l只需按顺序串联即可。核心原则是始终关闭上一个进程的stdout。import subprocess # 构建管道链 p1 subprocess.Popen([cat, file.txt], stdoutsubprocess.PIPE, textTrue) p2 subprocess.Popen([grep, error], stdinp1.stdout, stdoutsubprocess.PIPE, textTrue) p1.stdout.close() # p1 的输出已交给 p2父进程不再需要 p3 subprocess.Popen([wc, -l], stdinp2.stdout, stdoutsubprocess.PIPE, textTrue) p2.stdout.close() # p2 的输出已交给 p3父进程不再需要 # 获取最终输出 result, _ p3.communicate() print(f错误行数: {result.strip()}) # 确保所有进程都被回收如果它们还没退出 p1.wait() p2.wait() p3.wait()封装为通用函数便于复用import subprocess def safe_pipeline(commands): procs [] for i, cmd in enumerate(commands): if i 0: p subprocess.Popen(cmd, stdoutsubprocess.PIPE, textTrue) else: p subprocess.Popen(cmd, stdinprocs[-1].stdout, stdoutsubprocess.PIPE, textTrue) procs[-1].stdout.close() procs.append(p) output, _ procs[-1].communicate() # 等待所有进程结束并检查返回码 for p in procs: p.wait() if p.returncode ! 0: # 抛出异常或进行错误处理注意管道中 grep 无匹配时返回 1 是正常的 raise subprocess.CalledProcessError(p.returncode, p.args) return output try: # 故意让 grep 匹配不到会返回 1 result safe_pipeline([ [echo, abc], [grep, xyz] ]) except subprocess.CalledProcessError as e: print(f管道中的命令失败: {e})补充技巧同时获取 stderr如果你想分别获取每个命令的错误信息可以为每个Popen设置stderrsubprocess.PIPE然后通过p.communicate()或后续读取p.stderr来获取。3. 超时控制为run()或Popen.communicate()设置timeout参数可以防止子进程永久挂起-。try: subprocess.run([sleep, 10], timeout5) except subprocess.TimeoutExpired: print(命令执行超时)4. 环境变量与工作目录环境变量: 使用env参数可以为子进程设置独立的变量而不是继承父进程的。工作目录: 使用cwd参数可以指定子进程的执行路径。5. 错误处理设置checkTrue可以让run()在命令失败时抛出异常这比手动检查returncode更简洁。try: subprocess.run([false], checkTrue) except subprocess.CalledProcessError as e: print(f命令失败返回码: {e.returncode})⚠️ 避坑指南避免字符串拼接: 构建命令参数时始终使用列表而非字符串拼接以防止注入和参数解析错误。注意文本与字节: 使用textTrue处理文本使用bytes处理二进制数据。跨平台兼容性: Windows 和 Unix 的命令、路径风格不同编写脚本时需注意。例如Windows 上可执行文件通常有.exe后缀。僵尸进程: 使用Popen时务必调用proc.wait()或proc.communicate()来回收子进程资源避免产生僵尸进程-。PIPE死锁: 如果使用PIPE且数据量很大必须及时读取否则可能导致死锁。communicate()方法可以安全地处理这个问题-。import subprocess # 启动进程 process subprocess.Popen( [ping, -c, 5, google.com], stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, bufsize1 # 行缓冲 ) # 实时读取输出逐行 while True: output process.stdout.readline() if output and process.poll() is not None: break if output: print(output.strip()) # 获取剩余输出和返回码 stdout, stderr process.communicate()Popen作为上下文管理器: 从 Python 3.2 开始Popen支持上下文管理器可以确保资源被正确清理-。with subprocess.Popen([cmd], stdoutsubprocess.PIPE) as proc: print(proc.communicate()[0]) 总结使用subprocess模块记住以下三点优先使用subprocess.run()它足以应对 90% 的需求。默认使用参数列表除非绝对必要否则永远不要开启shellTrue。善用capture_output、text和check等参数编写更健壮的代码。参考资料https://docs.python.org/zh-cn/3/library/subprocess.html
