在进程python中运行函数时捕获真实的'stdout'和'stderr'

edqdpe6u  于 2023-02-02  发布在  Python
关注(0)|答案(1)|浏览(148)

我有一个python函数,想用multiprocessing包作为一个单独的进程运行它。

def run(ctx: Context):
    print("hello world!")
    return ctx

然后使用以下脚本将其作为单独的进程运行:

import multiprocessing

p = multiprocessing.Process(target=run, args=(ctx, ))
p.start()
p.join()

现在,我需要实时捕获上述进程的stdoutstderr,有没有类似这样的方法:

import subprocess
proc = subprocess.Popen(['python','fake_utility.py'],stdout=subprocess.PIPE)
while True:
  line = proc.stdout.readline()
  if not line:
    break

但是我需要传递不运行Popen命令的函数,当我在一个单独的进程中运行我的函数时,你知道我如何读取stdout吗?

nr7wwzry

nr7wwzry1#

我的方法是创建一个定制的上下文管理器,它可以用io.String()示例临时替换sys.stdoutsys.stderr,以捕获输出并返回它。为此,您需要使Processtarget 成为一个新函数,它可以设置上下文管理器并返回结果,为此使用了multiprocessing.Queue(顺便说一下,如果您希望run将其结果返回给主进程,则无论如何都需要这样做):

from multiprocessing import Process, Queue

from io import StringIO
import sys

class CaptureOutput:
    def __enter__(self):
        self._stdout_output = ''
        self._stderr_output = ''

        self._stdout = sys.stdout
        sys.stdout = StringIO()

        self._stderr = sys.stderr
        sys.stderr = StringIO()

        return self

    def __exit__(self, *args):
        self._stdout_output = sys.stdout.getvalue()
        sys.stdout = self._stdout

        self._stderr_output = sys.stderr.getvalue()
        sys.stderr = self._stderr

    def get_stdout(self):
        return self._stdout_output

    def get_stderr(self):
        return self._stderr_output

def run(ctx):
    print("hello world!")
    print("It works!", file=sys.stderr)
    raise Exception('Oh oh!') # Comment out to have a successful completion
    return ctx

def worker(ctx, queue):
    import traceback

    with CaptureOutput() as capturer:
        try:
            result = run(ctx)
        except Exception as e:
            result = e
            print(traceback.format_exc(), file=sys.stderr)
    queue.put((result, capturer.get_stdout(), capturer.get_stderr()))

if __name__ == '__main__':
    queue = Queue()
    ctx = None # for demo purposes
    p = Process(target=worker, args=(ctx, queue))
    p.start()
    # Must do this call before call to join:
    result, stdout_output, stderr_output = queue.get()
    p.join()
    print('stdout:', stdout_output)
    print('stderr:', stderr_output)

图纸:

stdout: hello world!

stderr: It works!
Traceback (most recent call last):
  File "C:\Booboo\test\test.py", line 44, in worker
    result = run(ctx)
  File "C:\Booboo\test\test.py", line 36, in run
    raise Exception('Oh oh!') # Comment out to have a successful completion
Exception: Oh oh!

相关问题