implement a detach() method for BufferedIOBase and TextIOBase #5883
diff --git a/Lib/_pyio.py b/Lib/_pyio.py
index e580366..e3e7c3d 100644
--- a/Lib/_pyio.py
+++ b/Lib/_pyio.py
@@ -642,6 +642,15 @@
"""
self._unsupported("write")
+ def detach(self) -> None:
+ """
+ Separate the underlying raw stream from the buffer and return it.
+
+ After the raw stream has been detached, the buffer is in an unusable
+ state.
+ """
+ self._unsupported("detach")
+
io.BufferedIOBase.register(BufferedIOBase)
@@ -689,13 +698,21 @@
self.raw.flush()
def close(self):
- if not self.closed:
+ if not self.closed and self.raw is not None:
try:
self.flush()
except IOError:
pass # If flush() fails, just give up
self.raw.close()
+ def detach(self):
+ if self.raw is None:
+ raise ValueError("raw stream already detached")
+ self.flush()
+ raw = self.raw
+ self.raw = None
+ return raw
+
### Inquiries ###
def seekable(self):
@@ -1236,6 +1253,15 @@
"""
self._unsupported("readline")
+ def detach(self) -> None:
+ """
+ Separate the underlying buffer from the TextIOBase and return it.
+
+ After the underlying buffer has been detached, the TextIO is in an
+ unusable state.
+ """
+ self._unsupported("detach")
+
@property
def encoding(self):
"""Subclasses should override."""
@@ -1448,11 +1474,12 @@
self._telling = self._seekable
def close(self):
- try:
- self.flush()
- except IOError:
- pass # If flush() fails, just give up
- self.buffer.close()
+ if self.buffer is not None:
+ try:
+ self.flush()
+ except IOError:
+ pass # If flush() fails, just give up
+ self.buffer.close()
@property
def closed(self):
@@ -1647,6 +1674,14 @@
self.seek(pos)
return self.buffer.truncate()
+ def detach(self):
+ if self.buffer is None:
+ raise ValueError("buffer is already detached")
+ self.flush()
+ buffer = self.buffer
+ self.buffer = None
+ return buffer
+
def seek(self, cookie, whence=0):
if self.closed:
raise ValueError("tell on closed file")
@@ -1865,3 +1900,7 @@
@property
def encoding(self):
return None
+
+ def detach(self):
+ # This doesn't make sense on StringIO.
+ self._unsupported("detach")
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
index a8c878e..1a525dc 100644
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -526,6 +526,12 @@
class CommonBufferedTests:
# Tests common to BufferedReader, BufferedWriter and BufferedRandom
+ def test_detach(self):
+ raw = self.MockRawIO()
+ buf = self.tp(raw)
+ self.assertIs(buf.detach(), raw)
+ self.assertRaises(ValueError, buf.detach)
+
def test_fileno(self):
rawio = self.MockRawIO()
bufio = self.tp(rawio)
@@ -811,6 +817,14 @@
bufio.flush()
self.assertEquals(b"".join(rawio._write_stack), b"abcghi")
+ def test_detach_flush(self):
+ raw = self.MockRawIO()
+ buf = self.tp(raw)
+ buf.write(b"howdy!")
+ self.assertFalse(raw._write_stack)
+ buf.detach()
+ self.assertEqual(raw._write_stack, [b"howdy!"])
+
def test_write(self):
# Write to the buffered IO but don't overflow the buffer.
writer = self.MockRawIO()
@@ -1052,6 +1066,10 @@
pair = self.tp(self.MockRawIO(), self.MockRawIO())
self.assertFalse(pair.closed)
+ def test_detach(self):
+ pair = self.tp(self.MockRawIO(), self.MockRawIO())
+ self.assertRaises(self.UnsupportedOperation, pair.detach)
+
def test_constructor_max_buffer_size_deprecation(self):
with support.check_warnings() as w:
warnings.simplefilter("always", DeprecationWarning)
@@ -1480,6 +1498,19 @@
self.assertRaises(TypeError, t.__init__, b, newline=42)
self.assertRaises(ValueError, t.__init__, b, newline='xyzzy')
+ def test_detach(self):
+ r = self.BytesIO()
+ b = self.BufferedWriter(r)
+ t = self.TextIOWrapper(b)
+ self.assertIs(t.detach(), b)
+
+ t = self.TextIOWrapper(b, encoding="ascii")
+ t.write("howdy")
+ self.assertFalse(r.getvalue())
+ t.detach()
+ self.assertEqual(r.getvalue(), b"howdy")
+ self.assertRaises(ValueError, t.detach)
+
def test_repr(self):
raw = self.BytesIO("hello".encode("utf-8"))
b = self.BufferedReader(raw)
diff --git a/Lib/test/test_memoryio.py b/Lib/test/test_memoryio.py
index ad04613..d94d9dd 100644
--- a/Lib/test/test_memoryio.py
+++ b/Lib/test/test_memoryio.py
@@ -57,6 +57,10 @@
class MemoryTestMixin:
+ def test_detach(self):
+ buf = self.ioclass()
+ self.assertRaises(self.UnsupportedOperation, buf.detach)
+
def write_ops(self, f, t):
self.assertEqual(f.write(t("blah.")), 5)
self.assertEqual(f.seek(0), 0)
@@ -336,6 +340,9 @@
class PyBytesIOTest(MemoryTestMixin, MemorySeekTestMixin, unittest.TestCase):
+
+ UnsupportedOperation = pyio.UnsupportedOperation
+
@staticmethod
def buftype(s):
return s.encode("ascii")
@@ -413,6 +420,7 @@
class PyStringIOTest(MemoryTestMixin, MemorySeekTestMixin, unittest.TestCase):
buftype = str
ioclass = pyio.StringIO
+ UnsupportedOperation = pyio.UnsupportedOperation
EOF = ""
# TextIO-specific behaviour.
@@ -518,9 +526,11 @@
class CBytesIOTest(PyBytesIOTest):
ioclass = io.BytesIO
+ UnsupportedOperation = io.UnsupportedOperation
class CStringIOTest(PyStringIOTest):
ioclass = io.StringIO
+ UnsupportedOperation = io.UnsupportedOperation
# XXX: For the Python version of io.StringIO, this is highly
# dependent on the encoding used for the underlying buffer.