Python：在迭代标准输出时从子进程中捕获异常

Question

我正在尝试从子进程中获取异常。如果我使用.communicate，我可以得到它，但我想避免使用它，因为我正在流式传输来自子进程的输出，并且不想等到整个子进程完成。还假设整个子流程可能需要很长时间。想知道如何在从子进程流式传输标准输出时捕获引发的异常。

考虑下面的例子，所以我想让版本 #1 工作，版本 #2 有点工作，但不希望那样。

在 main.py 中

import subprocess


class ExtProcess():
    def __init__(self, *args):
        self.proc = subprocess.Popen(['python', *args], stdout=subprocess.PIPE)

    def __iter__(self):
        return self

    def __next__(self):
        while True:
            line = self.proc.stdout.readline()
            if self.proc.returncode:
                raise Exception("error here")
            if not line:
                raise StopIteration
            return line


def run():
    ## version #1
    reader = ExtProcess("sample_extproc.py")
    for d in reader:
        print(f"got: {d}")

    ## version #2
    # proc = subprocess.Popen(['python', "sample_extproc.py"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    # output, error = proc.communicate()
    # print("got:", output)
    # if proc.returncode:
    #     raise Exception(error)

def main():
    try:
        print("start...")
        run()
        print("complete...")
    except Exception as e:
        print(f"Package midstream error here: {str(e)}")
    finally:
        print("clean up...")


if __name__ == "__main__":
    main()

在 sample_extproc.py 中

for x in range(10):
    print(x)
    if x == 3:
        raise RuntimeError("catch me")

我想从版本 #1 中获得类似以下内容的输出：

start...
got: b'0\r\n'
got: b'1\r\n'
got: b'2\r\n'
got: b'3\r\n'
Package midstream error here: b'Traceback (most recent call last):\r\n  File "sample_extproc.py", line 4, in <module>\r\n    raise RuntimeError("catch me")\r\nRuntimeError: catch me\r\n'
clean up...

基本上，它从子进程中遍历标准输出，然后在发生异常时打印异常，然后继续执行清理。

Answer 1

以下是我的问题的答案，完全基于@CharlesDuffy 的评论：

简而言之，确保在ExtProcess 类中有stderr=subprocess.PIPE，那么答案在版本#3，在迭代标准输出后，我们利用.wait() 和returncode 来检查是否有错误，如果是这样引发异常，则从stderr.read() 中获取错误以在父/主中捕获。

import subprocess

class ExtProcess():
    def __init__(self, *args):
        self.proc = subprocess.Popen(['python', *args], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

    def __iter__(self):
        return self

    def __next__(self):
        while True:
            line = self.proc.stdout.readline()
            if not line:
                raise StopIteration
            return line


def run():
    ## version #1
    # reader = ExtProcess("sample_extproc.py")
    # for d in reader:
    #     print(f"got: {d}")

    ## version #2
    # proc = subprocess.Popen(['python', "sample_extproc.py"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    # output, error = proc.communicate()
    # print("got:", output)
    # if proc.returncode:
    #     raise Exception(error)

    ## version #3
    reader = ExtProcess("sample_extproc.py")
    for d in reader:
        print(f"got: {d}")
    reader.proc.wait()
    if reader.proc.returncode:
       raise Exception(reader.proc.stderr.read())

def main():
    try:
        print("start...")
        run()
        print("complete...")
    except Exception as e:
        print(f"Package midstream error here: {str(e)}")
    finally:
        print("clean up...")


if __name__ == "__main__":
    main()