如何创建不会被关闭的sys.stdout副本?
问题描述
我想复制sys.stdout,让它不会被with语句关闭。现有一段通用代码:
def dosomething(fd: IO[bytes], input): with fd as f: for buf in input: f.write(buf)
我希望将sys.stdout传入该函数,且不会被with语句关闭。理想的实现方向如下:
class NoCloseIO(IO[bytes]): """IO[bytes] that does not close""" def close(self): pass dosomething(NoCloseIO(sys.stdout.buffer), io.StringIO("some text"))
请问该如何实现?是否可以通过reopen(sys.stdout)来实现?理想情况下,方案应适配任意TextIO或BinaryIO类型。
补充背景:我正在基于WebSocket协议实现自定义的subprocess.run接口,用于和远程进程交互。subprocess.run在stdout=PIPE或stdout=DEVNULL时会关闭文件描述符,但stdout=sys.stdout或stdout=1时不会关闭。我希望创建一个接口,把IO对象是否需要关闭的信息传递给WebSocket通信线程之上的解码线程。我可以给多个函数单独加变量,但更希望通过将特定场景下的close()设为无操作来简化代码。
解决方案
1. 通用无关闭IO包装类实现
可以编写一个通用包装类,代理任意IO对象(文本或二进制)的所有方法,仅将close方法替换为空操作,这样既保留原IO的全部功能,又能避免被with语句关闭:
import io from typing import IO, TypeVar T = TypeVar('T', bound=IO) class NoCloseIO: """包装任意IO对象,使其close()方法无操作""" def __init__(self, io_obj: T): self.io_obj = io_obj def close(self): # 空操作,不关闭原IO对象 pass # 代理原IO对象的所有其他属性与方法 def __getattr__(self, name): return getattr(self.io_obj, name)
用法示例
- 针对二进制IO(如
sys.stdout.buffer):dosomething(NoCloseIO(sys.stdout.buffer), io.BytesIO(b"some binary content")) - 针对文本IO(如
sys.stdout):def dosomething_text(fd: IO[str], input): with fd as f: for buf in input: f.write(buf) dosomething_text(NoCloseIO(sys.stdout), io.StringIO("some text content"))
这个类通过__getattr__自动代理原IO的所有方法和属性,无需为文本、二进制IO分别编写不同实现,完全适配你的通用需求。
2. 关于reopen(sys.stdout)的可行性
不推荐使用reopen方案:
reopen会基于文件描述符创建新的IO对象,和原sys.stdout并非同一实例,会丢失原对象的缓冲设置等状态;- 该方式仅适用于基于文件描述符的IO,无法适配内存IO(如
StringIO/BytesIO)等任意TextIO/BinaryIO类型,不符合你的通用要求。
3. 结合subprocess场景的优化
在你的自定义subprocess.run接口中,可以按以下逻辑处理:
- 当IO对象需要保留(如
sys.stdout)时,用NoCloseIO包装后传递给解码线程,避免线程执行关闭操作; - 当IO对象是临时资源(如
PIPE)时,直接传入原对象,让线程正常关闭释放资源。
这种方式无需给多个函数添加额外标记变量,通过IO对象自身的close行为传递是否需要关闭的信息,完美简化代码逻辑。
内容的提问来源于stack exchange,提问作者KamilCuk
相关产品推荐
相关产品推荐

