如何在Python异步函数中获取print输出的内容

同步环境下通过重定向sys.stdout到StringIO可以正常捕获函数的print输出，但给同步函数封装异步wrapper后，调用返回None，无法获取输出。

同步实现（正常工作） #

import sys
from io import StringIO

def sample():
    print('started to fetch log!')
    print('response:body')
    print('status:200')

def get_log():
    sys.stdout = StringIO()
    sample()
    log = sys.stdout.getvalue()
    sys.stdout = sys.__stdout__
    return log

# 同步调用
out = get_log()
print(out)
# 输出：
# started to fetch log!
# response:body
# status:200

有问题的异步封装 #

async def get():
    return asyncio.run(get_log()) 

问题原因 #

• 错误使用asyncio.run：asyncio.run用于启动新事件循环并运行异步函数，不能在已有事件循环的async def函数内部嵌套调用，会触发RuntimeError导致get_log未正常执行，最终返回None。
• 全局sys.stdout状态冲突：直接修改全局标准输出流，在异步/多线程环境下可能和事件循环的输出管理逻辑冲突，导致捕获失败。

解决方案 #

方案1：直接调用同步函数（短耗时场景） #

如果同步函数执行时间短，不会阻塞事件循环，可直接在异步函数内调用：

import asyncio
import sys
from io import StringIO

def sample():
    print('started to fetch log!')
    print('response:body')
    print('status:200')

def get_log():
    sys.stdout = StringIO()
    sample()
    log = sys.stdout.getvalue()
    sys.stdout = sys.__stdout__
    return log

async def get():
    return get_log()

# 调用异步函数
result = asyncio.run(get())
print(result)

方案2：线程池运行同步函数（长耗时场景） #

若同步函数执行时间长，为避免阻塞事件循环，使用线程池异步执行：

import asyncio
import sys
from io import StringIO
from concurrent.futures import ThreadPoolExecutor

def sample():
    print('started to fetch log!')
    print('response:body')
    print('status:200')

def get_log():
    sys.stdout = StringIO()
    sample()
    log = sys.stdout.getvalue()
    sys.stdout = sys.__stdout__
    return log

async def get():
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(ThreadPoolExecutor(), get_log)

# 调用测试
result = asyncio.run(get())
print(result)

方案3：避免修改全局sys.stdout（最健壮） #

修改函数逻辑，通过print的file参数指定输出流，彻底避免全局状态冲突：

import asyncio
from io import StringIO

def sample(output_stream):
    print('started to fetch log!', file=output_stream)
    print('response:body', file=output_stream)
    print('status:200', file=output_stream)

def get_log():
    output = StringIO()
    sample(output)
    return output.getvalue()

async def get():
    return get_log()

# 调用测试
result = asyncio.run(get())
print(result)

内容的提问来源于stack exchange，提问作者Mohammed Mohideen M Z