如何将stderr重定向至StreamWriter对象？

在Python中没有像Node.js那样直接的pipe方法，但可以通过两种常见场景实现类似效果：

场景1：捕获当前进程的stderr输出 #

Python的sys.stderr是同步文件对象，而StreamWriter是异步的，我们可以自定义一个重定向类，把同步的stderr写入操作转为异步写入到StreamWriter：

import sys
import asyncio
from asyncio import StreamWriter

class AsyncStderrRedirector:
    def __init__(self, writer: StreamWriter):
        self.writer = writer
        self.loop = asyncio.get_event_loop()

    def write(self, data):
        # 统一转为字节流（处理字符串/字节两种输入）
        if isinstance(data, str):
            data = data.encode('utf-8')
        # 异步提交写入任务到事件循环
        asyncio.run_coroutine_threadsafe(self.writer.write(data), self.loop)
    
    def flush(self):
        # 异步执行缓冲区刷新
        asyncio.run_coroutine_threadsafe(self.writer.drain(), self.loop)

# 使用示例
async def main():
    # 假设已创建StreamWriter实例（这里以连接本地服务为例）
    _, writer = await asyncio.open_connection('localhost', 8888)
    # 替换系统stderr
    sys.stderr = AsyncStderrRedirector(writer)
    
    # 测试输出到stderr
    print("测试错误信息", file=sys.stderr)
    
    # 确保数据写入完成后关闭连接
    await writer.drain()
    writer.close()
    await writer.wait_closed()

asyncio.run(main())

场景2：转发子进程的stderr到StreamWriter #

如果要把外部子进程的stderr输出导到StreamWriter，可以通过异步读取子进程的stderr流，再写入到目标StreamWriter：

import asyncio
from asyncio import StreamWriter

async def forward_stderr(subprocess_stderr, writer: StreamWriter):
    while chunk := await subprocess_stderr.read(1024):
        writer.write(chunk)
        await writer.drain()

async def main():
    # 创建StreamWriter实例
    _, writer = await asyncio.open_connection('localhost', 8888)
    
    # 启动会产生stderr输出的子进程
    proc = await asyncio.create_subprocess_exec(
        'ls', '-l', '/invalid/path',
        stderr=asyncio.subprocess.PIPE
    )
    
    # 转发子进程stderr到StreamWriter
    await forward_stderr(proc.stderr, writer)
    
    # 等待子进程结束并清理资源
    await proc.wait()
    writer.close()
    await writer.wait_closed()

asyncio.run(main())

注意事项 #

• 异步写入时要确保事件循环的一致性，多线程环境下必须用run_coroutine_threadsafe提交任务
• 注意编码统一，避免字符串转字节流时出现乱码
• 务必调用drain()方法，避免数据堆积在缓冲区中

内容的提问来源于stack exchange，提问作者vilal32