如何线程安全地捕获函数调用的 stdout 输出？

Question

我有一个程序现在将获得 REST API。我想我应该运行程序，捕获打印并将其放入消息响应中，而不是重写一堆代码。

关于如何从Python函数调用捕获stdout输出？我看到：

• 这个解决方案显然不是线程安全的。我在并发 future 中尝试过它，但我相信那里的评估，它不是线程安全的。
• 我还注意到了here的实现，但是虽然它使用线程是异步的，但它从未明确声明它本身是线程安全的。

这就是我在读到它显然不是线程安全之前尝试测试它是否是线程安全的方式。我从来没有遇到过任何问题（即输出始终是

['hello world', 'hello world2']

，但也许这是并发 future 的一个特征，在使用 REST API 模块（在我的例子中为 FastAPI）中的异步函数时不存在。

import concurrent.futures
from io import StringIO 
import sys

def main():
    num_tests = 30
    
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = [executor.submit(test) for _ in range(num_tests)]
        
        for future in concurrent.futures.as_completed(futures):
            try:
                result = future.result()
            except Exception as e:
                print(f"An error occurred: {e}")

class Capturing(list):
    def __enter__(self):
        self._stdout = sys.stdout
        sys.stdout = self._stringio = StringIO()
        return self
    def __exit__(self, *args):
        self.extend(self._stringio.getvalue().splitlines())
        del self._stringio    # free up some memory
        sys.stdout = self._stdout
def test():
    with Capturing() as output:
        print('hello world')

    print('displays on screen')

    with Capturing(output) as output:  # note the constructor argument
        print('hello world2')

    print('done')
    print('output:', output)
main()

在最好的情况下，我正在寻找如何异步捕获标准输出的想法。在最坏的情况下解释为什么这是不可能的。在这种情况下，我将寻找其他解决方案。

Answer 1

一种解决方案是使用

contextlib.redirect_stdout

:

import contextlib
 
def thread_proof_print_capture(thread_number):
    path = 'path/to/some/dir/thread_{thread_number}.txt'
    with open(path, 'w') as f:
        with contextlib.redirect_stdout(f):
            print('Hello, World')
    # the file thread_{thread_number}.txt contains all prints that would've been printed in this thread.

参见：

https://www.techiedelight.com/redirect-standard-output-to-file-python/