如何从 Python subprocess.check_output() 捕获异常输出？

Posted 2023-02-23

【中文标题】如何从 Python subprocess.check_output() 捕获异常输出？

【英文标题】：How to catch exception output from Python subprocess.check_output()?

【发布时间】：2014-09-11 01:36:38

【问题描述】：

我正在尝试从 Python 中进行比特币支付。在 bash 中，我通常会这样做：

bitcoin sendtoaddress <bitcoin address> <amount>

例如：

bitcoin sendtoaddress 1HoCUcbK9RbVnuaGQwiyaJGGAG6xrTPC9y 1.4214

如果成功，我会得到一个交易 ID 作为输出，但如果我尝试转移一个大于我的比特币余额的金额，我会得到以下输出：

error: "code":-4,"message":"Insufficient funds"

在我的 Python 程序中，我现在尝试按如下方式付款：

import subprocess

try:
    output = subprocess.check_output(['bitcoin', 'sendtoaddress', address, str(amount)])
except:
    print "Unexpected error:", sys.exc_info()

如果余额足够，它可以正常工作，但如果余额不足，sys.exc_info() 会打印出以下内容：

(<class 'subprocess.CalledProcessError'>, CalledProcessError(), <traceback object at 0x7f339599ac68>)

它不包括我在命令行上遇到的错误。所以我的问题是；如何从 Python 中获取输出的错误 ("code":-4,"message":"Insufficient funds")？

【参考方案1】：

根据subprocess.check_output() docs，错误引发的异常有一个output 属性，您可以使用它来访问错误详细信息：

try:
    subprocess.check_output(...)
except subprocess.CalledProcessError as e:
    print(e.output)

然后您应该能够分析此字符串并使用 json 模块解析错误详细信息：

if e.output.startswith('error: '):
    error = json.loads(e.output[7:]) # Skip "error: "
    print(error['code'])
    print(error['message'])

【讨论】：

我正在调用一个程序，它输出一些东西到 stdout 然后返回 1，但是 check_output 没有捕获它

@JorgeeFG 那我猜你的程序有问题。请注意，cmets 部分不适合提出新问题。如果您在特定问题上需要帮助，请点击页面右上角的“”大按钮。

在 Python 3.5+ 中，您可以使用 e.stderr 和 e.stdout 代替 e.output。

【参考方案2】：

我认为接受的解决方案无法处理在 stderr 上报告错误文本的情况。根据我的测试，异常的输出属性不包含来自 stderr 的结果，并且文档警告不要在 check_output() 中使用 stderr=PIPE。相反，我建议通过添加 stderr 支持对 J.F Sebastian 的解决方案进行小幅改进。毕竟，我们正在尝试处理错误，而 stderr 是它们经常被报告的地方。

from subprocess import Popen, PIPE

p = Popen(['bitcoin', 'sendtoaddress', ..], stdout=PIPE, stderr=PIPE)
output, error = p.communicate()
if p.returncode != 0: 
   print("bitcoin failed %d %s %s" % (p.returncode, output, error))

【讨论】：

我同意stderr 输出在这里非常相关。另一种解决方案是改用run() 函数（请参阅check_output docs 如何替换）。因为这样您就可以在错误报告中使用异常中的e.stderr。

这应该在顶部。

要保留输出的明显顺序，您可以使用stderr=STDOUT（合并两个流）。

我认为最好指出如果你不调用.communicate 那么.returncode 输出将为空（None）

if p.returncode != 0: 在这里不起作用，因为我得到None 成功。不得不使用if p.returncode:

【参考方案3】：

尝试“转移大于我的比特币余额的金额”并非意外错误。您可以直接使用Popen.communicate() 而不是check_output() 以避免不必要地引发异常：

from subprocess import Popen, PIPE

p = Popen(['bitcoin', 'sendtoaddress', ..], stdout=PIPE)
output = p.communicate()[0]
if p.returncode != 0: 
   print("bitcoin failed %d %s" % (p.returncode, output))

【讨论】：

Python 鼓励一种 EAFP 编程风格（请求宽恕比请求许可更容易），在这种情况下更喜欢异常处理而不是“if”检查。

@FerdinandBeyer：EAFP 不适用于这种情况：您不会拨打任何其他情况下不会拨打的电话。该代码没有 LBYL 结构：if check(): do()，您可以将其替换为 EAFP try: do() except Error: handle_error()。答案中的代码内联 check_output() 函数并避免在 if p.returncode 分支中引发异常，只是为了在同一级别捕获它。避免货物崇拜编程，想想

我们也可以这样做：p = Popen(['bitcoin', 'sendtoaddress', ..], stdout=PIPE, stderr=PIPE)，并将错误信息捕获为：output, error = p.communicate()。

对于使用管道的命令，我怎么能做到这一点？ @jfs

@alper 将命令作为字符串传递并添加 shell=True 参数：p = Popen("a | b", shell=True, ..)

【参考方案4】：

正如@Sebastian 所提到的，默认解决方案应该旨在使用run()： https://docs.python.org/3/library/subprocess.html#subprocess.run

这里是一个方便的实现（随意使用打印语句或您正在使用的任何其他日志记录功能更改日志类）：

import subprocess

def _run_command(command):
    log.debug("Command: ".format(command))
    result = subprocess.run(command, shell=True, capture_output=True)
    if result.stderr:
        raise subprocess.CalledProcessError(
                returncode = result.returncode,
                cmd = result.args,
                stderr = result.stderr
                )
    if result.stdout:
        log.debug("Command Result: ".format(result.stdout.decode('utf-8')))
    return result

以及示例用法（代码是不相关的，但我认为它可以作为一个示例，说明这个简单实现的可读性和处理错误是多么容易）：

try:
    # Unlock PIN Card
    _run_command(
        "sudo qmicli --device=/dev/cdc-wdm0 -p --uim-verify-pin=PIN1,"
        .format(pin)
    )

except subprocess.CalledProcessError as error:
    if "couldn't verify PIN" in error.stderr.decode("utf-8"):
        log.error(
                "SIM card could not be unlocked. "
                "Either the PIN is wrong or the card is not properly connected. "
                "Resetting module..."
                )
        _reset_4g_hat()
        return

【讨论】：

因为result.stderr包含字节，应该是if result.stderr.decode('utf-8')吗？

【参考方案5】：

这里有很好的答案，但在这些答案中，没有一个答案来自堆栈跟踪输出的文本，这是异常的默认行为。

如果您希望使用该格式化的回溯信息，您可能希望：

import traceback

try:
    check_call( args )
except CalledProcessError:
    tb = traceback.format_exc()
    tb = tb.replace(passwd, "******")
    print(tb)
    exit(1)

如您所知，如果您希望阻止显示的 check_call(args) 中有密码，上述内容很有用。

【讨论】：

注意记录秘密很重要^

【参考方案6】：

这对我有用。它捕获子进程的所有标准输出输出（对于 python 3.8）：

from subprocess import check_output, STDOUT
cmd = "Your Command goes here"
try:
    cmd_stdout = check_output(cmd, stderr=STDOUT, shell=True).decode()
except Exception as e:
    print(e.output.decode()) # print out the stdout messages up to the exception
    print(e) # To print out the exception message

【参考方案7】：

根据@macetw 的回答，我将异常直接打印到装饰器中的stderr。

Python 3

from functools import wraps
from sys import stderr
from traceback import format_exc
from typing import Callable, Collection, Any, Mapping


def force_error_output(func: Callable):
    @wraps(func)
    def forced_error_output(*args: Collection[Any], **kwargs: Mapping[str, Any]):
        nonlocal func

        try:
            func(*args, **kwargs)
        except Exception as exception:
            stderr.write(format_exc())
            stderr.write("\n")
            stderr.flush()

            raise exception

    return forced_error_output

Python 2

from functools import wraps
from sys import stderr
from traceback import format_exc


def force_error_output(func):
    @wraps(func)
    def forced_error_output(*args, **kwargs):
        try:
            func(*args, **kwargs)
        except Exception as exception:
            stderr.write(format_exc())
            stderr.write("\n")
            stderr.flush()

            raise exception

    return forced_error_output

然后在你的工人中使用装饰器

@force_error_output
def da_worker(arg1: int, arg2: str):
    pass

【参考方案8】：

从 Python 3.5 开始，subprocess.run() 支持 check 参数：

如果 check 为真，并且进程以非零退出代码退出，则会引发 CalledProcessError 异常。该异常的属性包含参数、退出代码以及标准输出和标准错误（如果它们被捕获）。

一个简单的例子，将引发并打印出CalledProcessError：

import subprocess
try:
    subprocess.run("exit 1", shell=True, check=True, timeout=15, capture_output=True)
except subprocess.CalledProcessError as e:
    print(e)  # Output: Command 'exit 1' returned non-zero exit status 1.