如何用Python subprocess实现多Shell管道的替代包装函数

我懂你想要的——一个不用依赖shell=True就能处理多管道命令的包装函数。虽然shell=True用起来省事，但它不仅有安全注入的风险，在跨平台场景下也容易出问题。刚好Python 2.7的subprocess文档里有个双管道的基础示例，我们可以把它扩展成支持任意多管道的版本。

Python 2.7 subprocess文档中给出的替代Shell管道的基础方案：

p1 = Popen(["dmesg"], stdout=PIPE)
p2 = Popen(["grep", "hda"], stdin=p1.stdout, stdout=PIPE)
p1.stdout.close() # Allow p1 to receive a SIGPIPE if p2 exits.
output = p2.communicate()[0]

多管道包装函数实现 #

from subprocess import Popen, PIPE

def run_piped_commands(commands):
    """
    执行多管道命令，无需依赖shell=True
    
    参数:
        commands: 嵌套列表结构，每个子列表对应一个命令及其参数
                  例如 [["dmesg"], ["grep", "hda"], ["head", "-n", "5"]] 对应 dmesg | grep hda | head -n 5
    
    返回:
        最终命令的输出结果（字符串格式）
    """
    if not commands:
        raise ValueError("至少需要传入一个命令")
    
    # 启动第一个进程
    prev_proc = Popen(commands[0], stdout=PIPE)
    
    # 逐个处理后续命令，依次连接管道
    for cmd in commands[1:]:
        curr_proc = Popen(cmd, stdin=prev_proc.stdout, stdout=PIPE)
        # 关闭前一个进程的stdout，确保它能在后续进程退出时收到SIGPIPE信号
        prev_proc.stdout.close()
        prev_proc = curr_proc
    
    # 获取最终输出，等待所有进程执行完毕
    final_output, _ = prev_proc.communicate()
    return final_output

使用示例 #

比如你要执行dmesg | grep hda | tail -n 3，只需要这样调用函数：

output = run_piped_commands([["dmesg"], ["grep", "hda"], ["tail", "-n", "3"]])
print(output)

关键细节说明 #

• 安全可靠：每个命令以列表形式传入，完全避免了shell命令注入的风险，同时跨平台兼容性更好
• 资源管理：每次创建新进程后关闭前一个进程的stdout，这是个关键操作——如果后续进程提前退出，前一个进程会收到SIGPIPE信号并终止，不会留下僵尸进程或者浪费资源
• 简化操作：函数封装了管道连接的逻辑，你只需要按顺序传入每个命令的参数列表即可

内容的提问来源于stack exchange，提问作者Kai Ye