:class:`BufferedIOBase` provides or overrides these methods in addition to
those from :class:`IOBase`:
+ .. method:: detach()
+
+ 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.
+
+ Some buffers, like :class:`BytesIO`, do not have the concept of a single
+ raw stream to return from this method. They raise
+ :exc:`UnsupportedOperation`.
+
.. method:: read([n])
Read and return up to *n* bytes. If the argument is omitted, ``None``, or
*max_buffer_size* is unused and deprecated.
- :class:`BufferedRWPair` implements all of :class:`BufferedIOBase`\'s methods.
+ :class:`BufferedRWPair` implements all of :class:`BufferedIOBase`\'s methods
+ except for :meth:`~BufferedIOBase.detach`, which raises
+ :exc:`UnsupportedOperation`.
.. class:: BufferedRandom(raw[, buffer_size[, max_buffer_size]])
A string, a tuple of strings, or ``None``, indicating the newlines
translated so far.
+ .. method:: detach()
+
+ Separate the underlying buffer from the :class:`TextIOBase` and return it.
+
+ After the underlying buffer has been detached, the :class:`TextIOBase` is
+ in an unusable state.
+
+ Some :class:`TextIOBase` implementations, like :class:`StringIO`, may not
+ have the concept of an underlying buffer and calling this method will
+ raise :exc:`UnsupportedOperation`.
+
.. method:: read(n)
Read and return at most *n* characters from the stream as a single
"""
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)
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):
"""
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."""
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):
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")
@property
def encoding(self):
return None
+
+ def detach(self):
+ # This doesn't make sense on StringIO.
+ self._unsupported("detach")
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)
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()
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)
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)
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)
class PyBytesIOTest(MemoryTestMixin, MemorySeekTestMixin, unittest.TestCase):
+
+ UnsupportedOperation = pyio.UnsupportedOperation
+
@staticmethod
def buftype(s):
return s.encode("ascii")
class PyStringIOTest(MemoryTestMixin, MemorySeekTestMixin, unittest.TestCase):
buftype = str
ioclass = pyio.StringIO
+ UnsupportedOperation = pyio.UnsupportedOperation
EOF = ""
# TextIO-specific behaviour.
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.
Core and Builtins
-----------------
+- Issue #5883: In the io module, the BufferedIOBase and TextIOBase ABCs have
+ received a new method, detach(). detach() disconnects the underlying stream
+ from the buffer or text IO and returns it.
+
- Issue #5859: Remove switch from '%f' to '%g'-style formatting for
floats with absolute value over 1e50. Also remove length
restrictions for float formatting: '%.67f' % 12.34 and '%.120e' %
return NULL;
}
+PyDoc_STRVAR(BufferedIOBase_detach_doc,
+ "Disconnect this buffer from its underlying raw stream and return it.\n"
+ "\n"
+ "After the raw stream has been detached, the buffer is in an unusable\n"
+ "state.\n");
+
+static PyObject *
+BufferedIOBase_detach(PyObject *self)
+{
+ return BufferedIOBase_unsupported("detach");
+}
+
PyDoc_STRVAR(BufferedIOBase_read_doc,
"Read and return up to n bytes.\n"
"\n"
static PyMethodDef BufferedIOBase_methods[] = {
+ {"detach", (PyCFunction)BufferedIOBase_detach, METH_NOARGS, BufferedIOBase_detach_doc},
{"read", BufferedIOBase_read, METH_VARARGS, BufferedIOBase_read_doc},
{"read1", BufferedIOBase_read1, METH_VARARGS, BufferedIOBase_read1_doc},
{"readinto", BufferedIOBase_readinto, METH_VARARGS, NULL},
PyObject *raw;
int ok; /* Initialized? */
+ int detached;
int readable;
int writable;
#define CHECK_INITIALIZED(self) \
if (self->ok <= 0) { \
- PyErr_SetString(PyExc_ValueError, \
- "I/O operation on uninitialized object"); \
+ if (self->detached) { \
+ PyErr_SetString(PyExc_ValueError, \
+ "raw stream has been detached"); \
+ } else { \
+ PyErr_SetString(PyExc_ValueError, \
+ "I/O operation on uninitialized object"); \
+ } \
return NULL; \
}
#define CHECK_INITIALIZED_INT(self) \
if (self->ok <= 0) { \
- PyErr_SetString(PyExc_ValueError, \
- "I/O operation on uninitialized object"); \
+ if (self->detached) { \
+ PyErr_SetString(PyExc_ValueError, \
+ "raw stream has been detached"); \
+ } else { \
+ PyErr_SetString(PyExc_ValueError, \
+ "I/O operation on uninitialized object"); \
+ } \
return -1; \
}
return res;
}
+/* detach */
+
+static PyObject *
+BufferedIOMixin_detach(BufferedObject *self, PyObject *args)
+{
+ PyObject *raw, *res;
+ CHECK_INITIALIZED(self)
+ res = PyObject_CallMethodObjArgs((PyObject *)self, _PyIO_str_flush, NULL);
+ if (res == NULL)
+ return NULL;
+ Py_DECREF(res);
+ raw = self->raw;
+ self->raw = NULL;
+ self->detached = 1;
+ self->ok = 0;
+ return raw;
+}
+
/* Inquiries */
static PyObject *
PyObject *raw;
self->ok = 0;
+ self->detached = 0;
if (!PyArg_ParseTupleAndKeywords(args, kwds, "O|n:BufferedReader", kwlist,
&raw, &buffer_size)) {
static PyMethodDef BufferedReader_methods[] = {
/* BufferedIOMixin methods */
+ {"detach", (PyCFunction)BufferedIOMixin_detach, METH_NOARGS},
{"flush", (PyCFunction)BufferedIOMixin_flush, METH_NOARGS},
{"close", (PyCFunction)BufferedIOMixin_close, METH_NOARGS},
{"seekable", (PyCFunction)BufferedIOMixin_seekable, METH_NOARGS},
PyObject *raw;
self->ok = 0;
+ self->detached = 0;
if (!PyArg_ParseTupleAndKeywords(args, kwds, "O|nn:BufferedReader", kwlist,
&raw, &buffer_size, &max_buffer_size)) {
static PyMethodDef BufferedWriter_methods[] = {
/* BufferedIOMixin methods */
{"close", (PyCFunction)BufferedIOMixin_close, METH_NOARGS},
+ {"detach", (PyCFunction)BufferedIOMixin_detach, METH_NOARGS},
{"seekable", (PyCFunction)BufferedIOMixin_seekable, METH_NOARGS},
{"readable", (PyCFunction)BufferedIOMixin_readable, METH_NOARGS},
{"writable", (PyCFunction)BufferedIOMixin_writable, METH_NOARGS},
PyObject *raw;
self->ok = 0;
+ self->detached = 0;
if (!PyArg_ParseTupleAndKeywords(args, kwds, "O|nn:BufferedReader", kwlist,
&raw, &buffer_size, &max_buffer_size)) {
static PyMethodDef BufferedRandom_methods[] = {
/* BufferedIOMixin methods */
{"close", (PyCFunction)BufferedIOMixin_close, METH_NOARGS},
+ {"detach", (PyCFunction)BufferedIOMixin_detach, METH_NOARGS},
{"seekable", (PyCFunction)BufferedIOMixin_seekable, METH_NOARGS},
{"readable", (PyCFunction)BufferedIOMixin_readable, METH_NOARGS},
{"writable", (PyCFunction)BufferedIOMixin_writable, METH_NOARGS},
return NULL;
}
+PyDoc_STRVAR(TextIOBase_detach_doc,
+ "Separate the underlying buffer from the TextIOBase and return it.\n"
+ "\n"
+ "After the underlying buffer has been detached, the TextIO is in an\n"
+ "unusable state.\n"
+ );
+
+static PyObject *
+TextIOBase_detach(PyObject *self)
+{
+ return _unsupported("detach");
+}
+
PyDoc_STRVAR(TextIOBase_read_doc,
"Read at most n characters from stream.\n"
"\n"
static PyMethodDef TextIOBase_methods[] = {
+ {"detach", (PyCFunction)TextIOBase_detach, METH_NOARGS, TextIOBase_detach_doc},
{"read", TextIOBase_read, METH_VARARGS, TextIOBase_read_doc},
{"readline", TextIOBase_readline, METH_VARARGS, TextIOBase_readline_doc},
{"write", TextIOBase_write, METH_VARARGS, TextIOBase_write_doc},
{
PyObject_HEAD
int ok; /* initialized? */
+ int detached;
Py_ssize_t chunk_size;
PyObject *buffer;
PyObject *encoding;
int r;
self->ok = 0;
+ self->detached = 0;
if (!PyArg_ParseTupleAndKeywords(args, kwds, "O|zzzi:fileio",
kwlist, &buffer, &encoding, &errors,
&newline, &line_buffering))
#define CHECK_INITIALIZED(self) \
if (self->ok <= 0) { \
- PyErr_SetString(PyExc_ValueError, \
- "I/O operation on uninitialized object"); \
+ if (self->detached) { \
+ PyErr_SetString(PyExc_ValueError, \
+ "underlying buffer has been detached"); \
+ } else { \
+ PyErr_SetString(PyExc_ValueError, \
+ "I/O operation on uninitialized object"); \
+ } \
return NULL; \
}
#define CHECK_INITIALIZED_INT(self) \
if (self->ok <= 0) { \
- PyErr_SetString(PyExc_ValueError, \
- "I/O operation on uninitialized object"); \
+ if (self->detached) { \
+ PyErr_SetString(PyExc_ValueError, \
+ "underlying buffer has been detached"); \
+ } else { \
+ PyErr_SetString(PyExc_ValueError, \
+ "I/O operation on uninitialized object"); \
+ } \
return -1; \
}
+static PyObject *
+TextIOWrapper_detach(PyTextIOWrapperObject *self)
+{
+ PyObject *buffer, *res;
+ CHECK_INITIALIZED(self);
+ res = PyObject_CallMethodObjArgs((PyObject *)self, _PyIO_str_flush, NULL);
+ if (res == NULL)
+ return NULL;
+ Py_DECREF(res);
+ buffer = self->buffer;
+ self->buffer = NULL;
+ self->detached = 1;
+ self->ok = 0;
+ return buffer;
+}
+
Py_LOCAL_INLINE(const Py_UNICODE *)
findchar(const Py_UNICODE *s, Py_ssize_t size, Py_UNICODE ch)
{
}
static PyMethodDef TextIOWrapper_methods[] = {
+ {"detach", (PyCFunction)TextIOWrapper_detach, METH_NOARGS},
{"write", (PyCFunction)TextIOWrapper_write, METH_VARARGS},
{"read", (PyCFunction)TextIOWrapper_read, METH_VARARGS},
{"readline", (PyCFunction)TextIOWrapper_readline, METH_VARARGS},
projects / python / commitdiff