I would want to copy sys.stdout such that it will not be closed.
There is some generic code that does:
def dosomething(fd: IO[bytes], input):
with fd as f:
for buf in input:
f.write(buf)
I would want to pass sys.stdout to it, in such a way, that it will not be closed with the with clause. Ideally:
class NoCloseIO(IO[bytes]):
"""IO[bytes] that does not close"""
def close(self):
pass
dosomething(NoCloseIO(sys.stdout.buffer), io.StringIO("some text"))
How would I do that? I do not understand how to do it. Can I reopen(sys.stdout) somehow?
Ideally I would want to do this with any TextIO or BinaryIO.
I see some confusion so let's add some color. I am doing my own subprocess.run interface over some protocol that uses websocket communication with a remote process. subprocess.run surely closes the file descriptor on stdout=PIPE or stdout=DEVNULL, but does not close when stdout=sys.stdout or stdout=1. I am trying to create an interface that will properly pass the information about if a IO-thing should be closed or not "down" the abstractions to a decoding thread above websocket communication thread. I could add a separate variable to many functions... or my intention is to just mark close() as a no-op in specific cases and keep the code clean.
If you cannot or do not want to alter the implementation of dosomething (monkey patching is often possible for non-native 3rd party code), you could write a wrapper class that intercepts the __exit__ method of the context management protocol:
import io
import sys
from typing import IO
def dosomething(fd: IO[bytes], input):
with fd as f:
for buf in input:
f.write(buf)
class NoCloseContextManager:
def __init__(self, obj):
self._wrapped_obj = obj
def __getattr__(self, attr):
# Allow the wrapped object's other methods to be accessed normally.
return getattr(self._wrapped_obj, attr)
def __enter__(self):
# Needed as both __dunder__ methods must exist directly in
# the class for it to be recognized as a context manager.
return self._wrapped_obj.__enter__()
def __exit__(self, exc_type, exc_value, traceback):
if exc_type is not None:
# Let exceptions get handled by the OG implementation.
return self._wrapped_obj.__exit__(exc_type, exc_value, traceback)
# Don't close anything otherwise.
dosomething(NoCloseContextManager(sys.stdout.buffer), io.BytesIO(b"some text"))
We can verify that the file descriptor is indeed not closed when the wrapper is in place:
buffer = sys.stdout.buffer
dosomething(buffer, io.BytesIO(b"text\n")) # Outputs: text
dosomething(buffer, io.BytesIO(b"more text\n")) # ValueError: I/O operation on closed file.
non_closing_buffer = NoCloseContextManager(sys.stdout.buffer)
dosomething(non_closing_buffer, io.BytesIO(b"text\n")) # Outputs: text
dosomething(non_closing_buffer, io.BytesIO(b"more text\n")) # Outputs: more text
Note that this unfortunately won't work as is if the wrapped filelike object is created as a context manager through using @contextlib.contextmanager on a generator function, which is a common pattern for user-created context managers:
import contextlib
class Printer:
def __init__(self):
print("Opening Printer")
def write(self, content):
print(content)
def close(self):
print("Closing Printer")
@contextlib.contextmanager
def context_managed_printer():
printer = Printer()
try:
yield printer
finally:
# `NoCloseContextManager` can't avoid calling this!
printer.close()
dosomething(context_managed_printer(), io.BytesIO(b"some text"))
dosomething(NoCloseContextManager(context_managed_printer()), io.BytesIO(b"some text")) # Still gets closed :(
Output:
Opening Printer b'some text' Closing Printer Opening Printer b'some text' Closing Printer
There also exists the standard library contextlib.nullcontext helper that simply returns the wrapped object as is from __enter__ and makes __exit__ a complete no-op: https://github.com/python/cpython/blob/7ce25edb8f41e527ed479bf61ef36dc9841b4ac5/Lib/contextlib.py#L775
For just passing sys.stdout.buffer to dosomething, it does work fine:
import contextlib
nullcontext_buffer = contextlib.nullcontext(sys.stdout.buffer)
dosomething(nullcontext_buffer, io.BytesIO(b"text\n")) # Outputs: text
dosomething(nullcontext_buffer, io.BytesIO(b"more text\n")) # Outputs: more text
And finally, if you really just want a copy of stdout, you can use os.dup to duplicate the file descriptor — then the context manager in dosomething won't close the original one:
import io
import os
import sys
duped_stdout = os.dup(sys.stdout.fileno())
with os.fdopen(duped_stdout, mode="wb") as buffer:
# Could also omit the with-block, as `dosomething` handles closing.
dosomething(buffer, io.BytesIO(b"text\n"))
print("Original stdout still open!")