subprocess 模块允许你生成新的进程,连接它们的输入、输出、错误管道,并且获取它们的返回码。此模块可以代替一些老旧的模块与功能,如os.system, os.spawn, os.popen()等。

1. Old Version Review

import os

os.popen('ls').read().split() # 可返回执行的结果
# ['data', 'preprocessing.ipynb', 'test.ipynb']

os.system('ls') # 只返回执行的退出状态
# 0

Better yet, you can use subprocess, it is safer, more powerful and likely faster.

2. subprocess.run()

subprocess过去版本中的call()check_call()check_output()已经被run()方法取代了。 subprocess.run()方法为3.5版本新增,是subprocess.Popen的简化版。大多数场景下,用run方法就足够了。对于更进阶的用例,也可以使用底层的 Popen 接口。

subprocess.run运行一个命令并等待命令结束返回结果,可以看做是自动调用Popencommunicate

1. 语法及部分参数说明
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)

功能:执行args参数所表示的命令,等待命令结束,并返回一个CompletedProcess类型对象,代表一个进程已经结束。

参数说明:

2. 使用例子
# example 1
import subprocess

result1 = subprocess.run('echo "I like python"', shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) # invoking shell
result2 = subprocess.run(['echo', 'I like potatos'], stdout=subprocess.PIPE, stderr=subprocess.PIPE) # without invoking shell
result3 = subprocess.run(['echo', 'I like potatos'], capture_output=True) # without invoking shell

print(result1)
print(result2)
print(result3)

# # output
# CompletedProcess(args='echo "I like python"', returncode=0, stdout=b'I like python\n', stderr=b'')
# CompletedProcess(args=['echo', 'I like potatos'], returncode=0, stdout=b'I like potatos\n', stderr=b'')
# CompletedProcess(args=['echo', 'I like potatos'], returncode=0, stdout=b'I like potatos\n', stderr=b'')
print(result2.returncode)
print(result2.stdout)
print(result2.stderr)
# # output
# 0
# b'I like potatos\n'
# b''
print("-"*20)
print(result3.returncode)
print(result3.stdout)
print(result3.stderr)
# # output
# 0
# b'I like potatos\n'
# None
# example 2
import subprocess
completed = subprocess.run(['ls', '-1'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
print('returncode:', completed.returncode)
print(f"结果的字节长度 {len(completed.stdout)}:\n{completed.stdout.decode('utf-8')}")
"""
returncode: 0
结果的字节长度 65:
classification.ipynb
hyperopt_output
regression.ipynb
test.ipynb
"""
# example 3, 错误处理
import subprocess
try:
    subprocess.run(['false'], check=True)
except subprocess.CalledProcessError as err:
    print('ERROR:', err)
# ERROR: Command '['false']' returned non-zero exit status 1.

false 命令总是以非零状态代码退出,run()将其解释为错误。 将 run()函数的 check 属性设置为 True,等同于使用check_returncode()方法。

3. 封装为易用的函数
import subprocess
import logging
logging.basicConfig(format='%(asctime)s : %(levelname)s : %(message)s',level=logging.INFO)

def bash2py(shell_command):
    """
    Args:
        shell_command: str, shell_command
    example:
        bash2py('du -sh')    
    """
    res = subprocess.run(shell_command, shell=True, capture_output=True)
    if res.returncode == 0:
        logging.info(f"Execute <{shell_command}> successfully!")
    else:
        raise Exception(f"ERROR: {res.stderr.decode('utf-8')}")
    return res.stdout.decode('utf-8').strip() # .split('\n')

3. subprocess.Popen()

Popen 是 subprocess的核心,子进程的创建和管理都靠它处理。一般场景下用run方法就可以了,Popen更加底层,灵活性更高。

语法:

class subprocess.Popen(args, bufsize=-1, executable=None, stdin=None, stdout=None, stderr=None, preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, universal_newlines=None, startupinfo=None, creationflags=0, restore_signals=True, start_new_session=False, pass_fds=(), *, encoding=None, errors=None, text=None)
import subprocess

result = subprocess.Popen(['echo', 'I like potatos'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)  
# without invoking shell
print(result)
# <subprocess.Popen object at 0x10e6df7f0>
out, err = result.communicate()
print(out, err)
# b'I like potatos\n' b''
print(f"STDOUT: {out.decode('ascii')}STDERR: {err.decode('ascii')}")
# STDOUT: I like potatos
# STDERR:

REFERENCE

subprocess

subprocess-example

Python 调用系统命令的模块 Subprocess