代码收藏家技术教程 2024-10-20

Python之subprocess模块

subprocess模块概述

旧API函数简介

subprocess.run()函数介绍

语法

参数说明

返回值

使用示例

运行外部应用程序

运行Shell命令

传递输入到子进程

使用其他参数

处理子进程的错误

处理命令超时

subprocess异常处理

1. subprocess.CalledProcessError

2. subprocess.TimeoutExpired

3. FileNotFoundError

4. 综合示例

与 Shell 交互的 subprocess 模块简介

1. 基于 UNIX 的 Shell 使用

2. Windows Shell 使用

3. 安全警告

总结

与进程通信

标准 I/O 流

标准流的解码

进程间通信的实用技巧

subprocess 模块中使用管道

使用管道连接进程

在不同平台上使用管道

Windows 上的管道

类 UNIX 系统上的管道

使用管道进行复杂任务

管道的注意事项

subprocess模块概述

subprocess模块是Python中用于启动和管理子进程的一个强大工具，替代了旧有的os.system和os.spawn系列函数。通过这个模块，您可以从Python脚本中执行系统命令和其他可执行程序，并与之进行交互。
在Python的subprocess模块中，有一些旧的API函数，比如call()、check_call()和check_output()。这些函数在Python 3.5及更早版本中被广泛使用。虽然这些函数仍然可以使用，但从Python 3.5开始，subprocess.run()函数被引入，作为更统一和灵活的接口来替代它们。
尽管旧API依然存在以支持向后兼容性，但在新项目中推荐使用subprocess.run()，因为它提供了更一致的参数和更丰富的功能。

旧API函数简介

以下是旧API函数的基本介绍及其与subprocess.run()的对应关系：

subprocess.call()
subprocess.call()用于执行命令并等待它完成，返回命令的退出状态。

import subprocess

# 使用call()
return_code = subprocess.call(['ls', '-l'])
print('Return code:', return_code)

call()的等效run()版本：

import subprocess

# 使用run()
result = subprocess.run(['ls', '-l'])
print('Return code:', result.returncode)

subprocess.check_call()
subprocess.check_call()与call()类似，但在命令返回非零退出状态时会引发subprocess.CalledProcessError异常。

import subprocess

try:
    subprocess.check_call(['ls', '-l'])
except subprocess.CalledProcessError as e:
    print(f'Command failed with return code: {e.returncode}')

check_call()的等效run()版本：

import subprocess
try:
    subprocess.run(['ls', '-l'], check=True)
except subprocess.CalledProcessError as e:
    print(f'Command failed with return code: {e.returncode}')
subprocess.check_output()

subprocess.check_output()用于执行命令并返回其输出。如果命令返回非零退出状态，将引发subprocess.CalledProcessError异常。

import subprocess

try:
    output = subprocess.check_output(['ls', '-l'])
    print('Command output:', output.decode())
except subprocess.CalledProcessError as e:
    print(f'Command failed with return code: {e.returncode}')

check_output()的等效run()版本：

import subprocess

try:
    result = subprocess.run(['ls', '-l'], capture_output=True, text=True, check=True)
    print('Command output:', result.stdout)
except subprocess.CalledProcessError as e:
    print(f'Command failed with return code: {e.returncode}')

subprocess.run()的优势
更一致的参数接口：run()的参数设计更直观，比如capture_output、text等。
更丰富的返回信息：返回一个CompletedProcess对象，包含stdout、stderr和returncode等信息。
统一的错误处理机制：通过check参数即可处理错误情况。

subprocess.run()函数介绍

语法

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

参数说明

args：这是必须的参数，用于指定要执行的命令。可以是字符串（如果shell=True）或列表。列表的第一个元素通常是命令名称，后续元素是参数。例如，[‘ls’,‘-l’]。

stdin、stdout、stderr：用于设置标准输入、输出和错误的流，可以是subprocess.PIPE、subprocess.DEVNULL等。默认情况下，这些流不会被捕获。

capture_output：如果为True，则同时捕获stdout和stderr，相当于设置stdout=subprocess.PIPE和stderr=subprocess.PIPE。

input ：input参数是可选的，用于向子进程的标准输入传递数据。可以传递字节 (bytes) 或字符串 (str) 数据给 input 参数。
当 text=True 或 universal_newlines=True 时，input 应为字符串 (str) 类型；否则，它应为字节 (bytes) 类型。
使用 input 参数时，不应同时使用 Popen 构造函数的 stdin 参数，因为它会在内部被使用。传递给 input 的数据将被子进程读取，例如通过 sys.stdin.read()。

shell：如果为True，命令将在shell中执行（例如，/bin/sh）。这对于需要shell特性（如通配符、管道等）的命令非常有用，但也存在一定的安全风险，特别是在处理不受信任的输入时。

cwd：指定执行命令时的当前工作目录。

timeout：如果命令执行时间超过此秒数，则抛出TimeoutExpired异常。

check：如果为True，并且命令返回非零退出状态，则抛出CalledProcessError异常。

encoding和errors：用于设置输出解码时使用的编码和错误处理方式。

text或universal_newlines：如果为True，则以文本模式打开stdout和stderr，将其解码为字符串。这是将encoding和errors结合在一起的快捷方式。

env：用于指定子进程的环境变量。可以传递一个字典来设置环境变量。

返回值

CompletedProcess对象,该对象包含以下属性：

args：执行的命令和参数。

returncode：命令的退出状态码。0表示成功，非零表示失败。

stdout：捕获的标准输出（如果设置为捕获）。

stderr：捕获的标准错误输出（如果设置为捕获）。

使用示例

运行外部应用程序

要运行一个可执行文件（如.exe文件），你需要提供程序的路径和必要的参数。
示例：运行外部应用程序
假设我们要运行一个名为 example.exe 的程序，并传递一些参数：

import subprocess

# 完整的可执行文件路径
executable_path = r'C:\Path\To\Your\Executable.exe'

# 执行可执行文件
result = subprocess.run(
    [executable_path, 'arg1', 'arg2'],  # 替换为实际的参数
    capture_output=True,  # 捕获输出
    text=True  # 将输出解码为文本
)

# 检查是否成功
if result.returncode == 0:
    print("应用程序成功执行，输出如下：")
    print(result.stdout)
else:
    print("应用程序执行失败，错误如下：")
    print(result.stderr)

又如打开一个文本编辑器窗口

>>> subprocess.run(["notepad"])
CompletedProcess(args=['notepad'], returncode=0)

运行Shell命令

如果你需要运行一个操作系统的内置命令（如dir、ls、echo等），可以使用 shell=True 参数。这会让命令在 shell 中执行。
示例：在 Windows 上运行 dir 命令

import subprocess

# 使用 shell 执行 Windows 内置命令
result = subprocess.run(
    'dir',  # 直接作为字符串传入命令
    shell=True,  # 使用 shell 执行
    capture_output=True,
    text=True
)

# 打印结果
if result.returncode == 0:
    print("命令成功执行，输出如下：")
    print(result.stdout)
else:
    print("命令执行失败，错误如下：")
    print(result.stderr)

示例：在 Linux 或 macOS 上运行 ls 命令

import subprocess

# 使用 shell 执行 Linux/macOS 内置命令
result = subprocess.run(
    'ls -l',  # 直接作为字符串传入命令
    shell=True, 
    capture_output=True,
    text=True
)

# 打印结果
if result.returncode == 0:
    print("命令成功执行，输出如下：")
    print(result.stdout)
else:
    print("命令执行失败，错误如下：")
    print(result.stderr)

传递输入到子进程

有时你可能需要向子进程的标准输入提供输入数据，可以通过 input 参数实现。

示例：向子进程传递输入

import subprocess

# 向子进程传递输入
result = subprocess.run(
    ['python', '-c', 'print(input())'],  # 子进程执行Python命令
    input='Hello, World!',  # 传递的输入数据
    capture_output=True,
    text=True
)

# 打印子进程的输出
if result.returncode == 0:
    print("子进程输出：")
    print(result.stdout)
else:
    print("子进程执行失败，错误如下：")
    print(result.stderr)

使用其他参数

subprocess.run() 函数还支持许多其他参数，可以让你更加灵活地管理子进程。以下是一些常用参数：

cwd: 指定子进程的工作目录。
timeout: 设置子进程的超时时间。
env: 指定子进程使用的环境变量。
check: 如果设置为True，子进程返回非零退出码时会引发 CalledProcessError 异常。
示例：使用其他参数

import subprocess

# 设置工作目录和环境变量
result = subprocess.run(
    ['python', '-c', 'import os; print(os.getcwd()); print(os.getenv("MY_VAR"))'],
    cwd=r'C:\Path\To\Working\Directory',  # 工作目录
    env={'MY_VAR': 'Hello, Environment!'},  # 环境变量
    capture_output=True,
    text=True,
    check=True  # 检查返回码
)

# 打印结果
print("子进程输出：")
print(result.stdout)

处理子进程的错误

你可以通过捕获 subprocess.CalledProcessError 异常来处理子进程中的错误。

示例：处理子进程错误

import subprocess

try:
    # 尝试运行一个可能失败的命令
    result = subprocess.run(
        ['python', '-c', 'exit(1)'],  # 这个命令将返回1，表示失败
        capture_output=True,
        text=True,
        check=True  # 启用错误检查
    )
except subprocess.CalledProcessError as e:
    # 捕获异常并打印错误信息
    print(f"子进程执行失败，错误码：{e.returncode}")
    print(f"错误输出：{e.stderr}")

处理命令超时

import subprocess

try:
    # 设置超时
    subprocess.run(['sleep', '10'], timeout=5)
except subprocess.TimeoutExpired:
    print('The command timed out!')

subprocess异常处理

1. subprocess.CalledProcessError

当子进程返回一个非零的退出状态码时，会引发 subprocess.CalledProcessError。这表明子进程失败了。你可以通过传递 check=True 参数来启用此行为。

示例代码：

import subprocess

try:
    result = subprocess.run(
        ["python", "timer.py"],
        check=True
    )
except subprocess.CalledProcessError as exc:
    print(f"进程失败，因为返回了非零退出状态码: {exc.returncode}")
    print(f"错误信息: {exc.stderr}")

2. subprocess.TimeoutExpired

如果子进程执行超过了指定的时间限制，会引发 subprocess.TimeoutExpired。可以通过 timeout 参数设置时间限制。
示例代码：

import subprocess

try:
    subprocess.run(
        ["python", "timer.py", "5"],
        timeout=1
    )
except subprocess.TimeoutExpired as exc:
    print(f"进程超时，执行超过了 {exc.timeout} 秒")
    print(f"错误信息: {exc}")

3. FileNotFoundError

如果指定的程序不存在，系统会引发 FileNotFoundError。这是最常见的错误之一，当你尝试执行一个不存在的文件时会遇到。
示例代码：

import subprocess

try:
    subprocess.run(["now_you_see_me"])
except FileNotFoundError as exc:
    print(f"进程失败，因为指定的可执行文件未找到。")
    print(f"错误信息: {exc}")

4. 综合示例

以下代码展示了如何处理上述三种异常：

import subprocess

try:
    subprocess.run(
        ["python", "timer.py", "5"],
        timeout=10,  # 设置超时时间为10秒
        check=True   # 当子进程返回非零退出状态码时引发异常
    )
except FileNotFoundError as exc:
    print(f"进程失败，因为可执行文件未找到。\n{exc}")
except subprocess.CalledProcessError as exc:
    print(f"进程失败，因为返回了非零退出码。返回值: {exc.returncode}\n{exc}")
except subprocess.TimeoutExpired as exc:
    print(f"进程超时。\n{exc}")

与 Shell 交互的 subprocess 模块简介

在使用 Python 的 subprocess 模块时，有时需要与基于文本的程序（如 Shell 命令）进行交互。理解如何正确地与这些程序交互可以帮助你更高效地执行系统命令和脚本。以下是一些与 Shell 交互的常见用法和注意事项。

1. 基于 UNIX 的 Shell 使用

在 UNIX 系统（如 Linux 和 macOS）中，Shell 是与系统交互的主要工具。你可以使用 subprocess.run() 直接调用 Shell 命令。可以使用 shell=True 参数来运行 Shell 命令，或者显式调用 Shell 并传递命令。

示例：

import subprocess

# 直接使用 shell=True
result = subprocess.run("ls /usr/bin | grep pycode", shell=True, text=True)
print(result.stdout)

# 使用显式调用 bash
result = subprocess.run(["bash", "-c", "ls /usr/bin | grep pycode"], text=True)
print(result.stdout)

在 UNIX 系统中，sh 通常是 Bourne shell 的符号链接，而 bash 或 zsh 是更常用的 Shell。-c 参数告诉 Shell 执行接下来的命令。

2. Windows Shell 使用

在 Windows 中，你可以使用 cmd 或 PowerShell 来运行命令。cmd 是传统的命令提示符，而 PowerShell 提供了更强大的功能。

示例：

import subprocess

# 使用 PowerShell
result = subprocess.run(["pwsh", "-Command", "Get-ChildItem C:\\RealPython"], text=True)
print(result.stdout)

# 使用 cmd
result = subprocess.run(["cmd", "/c", "dir C:\\RealPython"], text=True)
print(result.stdout)

在 Windows 中，/c 参数用于 cmd，它告诉 cmd 执行指定的命令后退出。-Command 参数用于 PowerShell，它告诉 PowerShell 执行后面的命令。

3. 安全警告

当从用户输入构建命令时，必须小心注入攻击。攻击者可以利用未过滤的输入执行恶意命令。这是一个常见的安全问题，尤其是在处理来自不可信来源的输入时。

不安全的示例：

import subprocess

# 从用户获取输入并直接传递给 Shell
user_input = input("Enter directory to list: ")
subprocess.run(["pwsh", "-Command", f"Get-ChildItem {user_input}"], text=True)

安全的做法：

避免直接拼接用户输入到命令中。
使用列表格式传递参数，而不是通过字符串传递。
改进示例：

import subprocess

# 从用户获取输入并验证
user_input = input("Enter directory to list: ")
safe_input = user_input.replace(";", "").replace("&", "").replace("|", "")  # 移除危险字符
subprocess.run(["pwsh", "-Command", f"Get-ChildItem {safe_input}"], text=True)

这种处理方式可以减少恶意输入的风险，但仍需注意可能存在的其他安全漏洞。

总结

基于 UNIX 的 Shell：可以直接调用 Shell 或通过 shell=True 参数执行 Shell 命令。

WindowsShell：使用 cmd 或 PowerShell，根据需要选择相应的命令和参数。

安全问题：在处理用户输入时要小心，避免命令注入攻击。

与进程通信

在使用 subprocess 模块时，理解如何与进程进行通信是非常重要的。这包括如何读取和写入进程的标准输入（stdin）、标准输出（stdout）和标准错误（stderr）流。以下是一些关键概念和示例，帮助你更好地理解这些操作。

标准 I/O 流

在进程中，标准 I/O 流用于进程之间的通信：

stdin：用于从外部接收输入。

stdout：用于向外部发送标准输出。

stderr：用于向外部发送错误消息。

这些流是进程间通信的基础，通常用于处理命令行工具和其他基于文本的程序。

基本示例

捕获子进程的输出
你可以使用 capture_output=True 参数来捕获子进程的标准输出和标准错误输出。输出将作为字节对象存储在 CompletedProcess 对象的 stdout 和 stderr 属性中。

示例：

import subprocess

# 假设 magic_number.py 输出一个随机数
magic_number_process = subprocess.run(
    ["python", "magic_number.py"], capture_output=True
)
print(magic_number_process.stdout)  # 输出以字节对象形式显示

要将字节对象转换为字符串，可以使用 decode() 方法或通过指定 encoding 参数。

示例（使用编码）：

magic_number_process = subprocess.run(
    ["python", "magic_number.py"], capture_output=True, encoding="utf-8"
)
print(magic_number_process.stdout)  # 输出以字符串形式显示

向子进程写入输入
你可以使用 input 参数向子进程的标准输入流写入数据。这对于需要用户输入的程序特别有用。

示例：

import subprocess

# 向反应游戏程序提供两个换行符
process = subprocess.run(
    ["python", "reaction_game.py"], input="\n\n", encoding="utf-8"
)
print(process.stdout)  # 打印游戏输出

在这个示例中，我们通过 input 参数传递了两个换行符，这将模拟用户的按键操作。

标准流的解码

进程的通信以字节为单位进行。为了将这些字节转换为文本，你可以使用以下方法：

文本模式：通过设置 text=True 或 encoding 参数，subprocess.run() 可以自动处理编码和解码。
示例（文本模式）：

result = subprocess.run(
    ["python", "magic_number.py"], capture_output=True, text=True
)
print(result.stdout)  # 输出以字符串形式显示

显式解码：使用 decode() 方法将字节对象转换为字符串。
示例（显式解码）：

result = subprocess.run(
    ["python", "magic_number.py"], capture_output=True
)
output = result.stdout.decode("utf-8")
print(output)  # 输出以字符串形式显示

进程间通信的实用技巧

安全性：使用 shell=True 时要小心，尤其是处理用户输入时，防止命令注入攻击。

错误处理：通过 stderr 捕获错误输出，并根据需要处理。

示例（捕获错误）：

result = subprocess.run(
    ["python", "non_existent_script.py"], capture_output=True, text=True
)
print("Error:", result.stderr)  # 输出错误信息

流的同步：子进程的标准输入、输出和错误流可能会被缓冲。可以使用 flush=True 参数强制刷新缓冲区。
示例（强制刷新）：

import subprocess
import sys

# 向子进程的标准输入流写入数据
process = subprocess.Popen(
    ["python", "reaction_game.py"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 向标准输入流写入数据
process.stdin.write("\n\n")
process.stdin.flush()

# 读取标准输出流
output = process.stdout.read()
print(output)

subprocess 模块中使用管道

在 Python 中，subprocess 模块提供了创建和管理子进程的功能，可以通过管道连接多个进程以实现更复杂的数据流处理。管道允许你将一个进程的输出直接作为另一个进程的输入，从而在进程之间进行数据传递。

在本节中，我们将详细介绍如何在 subprocess 模块中使用管道，以及如何将它们应用于实际问题中。

使用管道连接进程

管道在 subprocess 中的实现依赖于 Popen 类，它允许你同时启动多个进程并连接它们的输入和输出。下面是一个使用管道将两个进程连接在一起的例子。

示例：使用管道将两个进程连接
假设我们要在 /usr/bin 目录中查找包含 python 字符串的文件。我们可以使用 ls 列出目录内容，然后使用 grep 过滤结果。

import subprocess

# 启动第一个子进程，列出目录内容
ls_process = subprocess.Popen(
    ["ls", "/usr/bin"],
    stdout=subprocess.PIPE,  # 将 stdout 设置为管道
)

# 启动第二个子进程，从第一个进程的输出中读取数据
grep_process = subprocess.Popen(
    ["grep", "python"],
    stdin=ls_process.stdout,  # 将 stdin 设置为第一个进程的输出
    stdout=subprocess.PIPE,   # 将 stdout 设置为管道
)

# 关闭第一个进程的输出流，以便 grep 进程能接收到 EOF
ls_process.stdout.close()

# 从第二个进程读取结果
output, _ = grep_process.communicate()

# 打印结果
print(output.decode())

代码说明

Popen 类：使用 Popen 启动子进程，并通过管道连接它们的输入和输出。

管道设置：在 ls_process 中，将stdout 设置为 subprocess.PIPE，表示其输出被重定向到一个管道。在 grep_process 中，将 stdin设置为第一个进程的 stdout，将 grep 的输入连接到 ls 的输出。

关闭输出流： ls_process.stdout.close() 用于通知 grep 进程读取结束，从而避免死锁。

结果读取：使用 communicate() 方法从grep_process中读取输出。这会同时等待进程完成并返回它的输出。

这种方式允许你将多个命令串联起来，就像在 shell 中使用管道符 (|) 一样。

在不同平台上使用管道

使用 subprocess 和管道时需要考虑平台差异，特别是在 Windows 和类 UNIX 系统之间。管道的使用在这些平台上是类似的，但需要注意一些特定细节：

Windows 上的管道

在 Windows 上，管道的工作方式与类 UNIX 系统类似。你可以使用 Popen 和 PIPE 来将进程连接在一起。然而，某些 shell 特定的命令可能需要使用 shell=True 来启用 shell 支持。

import subprocess

# 使用 shell=True 在 Windows 上运行 shell 命令
process = subprocess.Popen(
    "dir | findstr python",
    shell=True,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)

output, error = process.communicate()

print(output.decode())

类 UNIX 系统上的管道

在类 UNIX 系统上，如 Linux 和 macOS，你可以直接使用 Popen 和管道来连接进程，通常无需 shell=True，除非你需要使用 shell 特定的功能。

import subprocess

# 启动第一个进程，列出目录内容
ls_process = subprocess.Popen(
    ["ls", "/usr/bin"],
    stdout=subprocess.PIPE,  # 将 stdout 设置为管道
)

# 启动第二个进程，从第一个进程的输出中读取数据
grep_process = subprocess.Popen(
    ["grep", "python"],
    stdin=ls_process.stdout,  # 将 stdin 设置为第一个进程的输出
    stdout=subprocess.PIPE,   # 将 stdout 设置为管道
)

# 关闭第一个进程的输出流，以便 grep 进程能接收到 EOF
ls_process.stdout.close()

# 从第二个进程读取结果
output, _ = grep_process.communicate()

# 打印结果
print(output.decode())

使用管道进行复杂任务

管道不仅限于简单的命令连接，还可以用于更复杂的数据流处理任务。例如，你可以使用管道将多个数据处理步骤串联在一起，实现数据的流水线式处理。

示例：使用管道处理数据
假设我们有一个文本文件，其中包含多行文本。我们希望对文本进行过滤，然后将结果排序输出。我们可以使用 subprocess 和管道来实现这一点：

import subprocess

# 创建一个子进程进行过滤
grep_process = subprocess.Popen(
    ["grep", "pattern", "input.txt"],  # 从 input.txt 中查找包含 pattern 的行
    stdout=subprocess.PIPE,            # 将输出设置为管道
)

# 创建一个子进程进行排序
sort_process = subprocess.Popen(
    ["sort"],                          # 对 grep 的输出进行排序
    stdin=grep_process.stdout,         # 将 grep 的输出作为输入
    stdout=subprocess.PIPE,            # 将输出设置为管道
)

# 关闭 grep 进程的输出流
grep_process.stdout.close()

# 从排序进程中读取结果
output, _ = sort_process.communicate()

# 打印结果
print(output.decode())

管道的注意事项

在使用管道时，需要注意以下几点：

死锁问题：如果一个进程的输出缓冲区被填满，而下一个进程没有及时读取，可能会导致死锁。确保及时读取输出，并关闭不必要的文件描述符。

数据量限制：管道的缓冲区大小是有限的。在处理大数据量时，可能需要考虑分块处理，或使用文件替代管道。

错误处理：使用 subprocess.CalledProcessError 捕获命令执行错误，使用 try-except 结构处理异常。

平台差异：不同平台可能对命令和管道的处理有所不同，确保代码在目标平台上经过测试。

作者：船长的日记

物联沃分享整理
物联沃-IOTWORD物联网 » Python之subprocess模块

代码收藏家普通

分享到：

Python之subprocess模块

subprocess模块概述

旧API函数简介

subprocess.run()函数介绍

语法

参数说明

返回值

使用示例

运行外部应用程序

运行Shell命令

传递输入到子进程

使用其他参数

处理子进程的错误

处理命令超时

subprocess异常处理

1. subprocess.CalledProcessError

2. subprocess.TimeoutExpired

3. FileNotFoundError

4. 综合示例

与 Shell 交互的 subprocess 模块简介

1. 基于 UNIX 的 Shell 使用

2. Windows Shell 使用

3. 安全警告

总结

与进程通信

标准 I/O 流

标准流的解码

进程间通信的实用技巧

subprocess 模块中使用管道

使用管道连接进程

在不同平台上使用管道

Windows 上的管道

类 UNIX 系统上的管道

使用管道进行复杂任务

管道的注意事项

代码收藏家 普通

相关推荐

发表回复 取消回复

代码收藏家普通

发表回复取消回复