Whether line buffering is enabled.
+ .. attribute:: write_through
+
+ Whether writes are passed immediately to the underlying binary
+ buffer.
+
+ .. versionadded:: 3.7
+
+ .. method:: reconfigure(*, line_buffering=None, write_through=None)
+
+ Reconfigure this text stream using new settings for *line_buffering*
+ and *write_through*. Passing ``None`` as an argument will retain
+ the current setting for that parameter.
+
+ This method does an implicit stream flush before setting the
+ new parameters.
+
+ .. versionadded:: 3.7
+
.. class:: StringIO(initial_value='', newline='\\n')
raise ValueError("invalid errors: %r" % errors)
self._buffer = buffer
- self._line_buffering = line_buffering
self._encoding = encoding
self._errors = errors
self._readuniversal = not newline
# Sometimes the encoder doesn't exist
pass
+ self._configure(line_buffering, write_through)
+
+ def _configure(self, line_buffering=False, write_through=False):
+ self._line_buffering = line_buffering
+ self._write_through = write_through
+
# self._snapshot is either None, or a tuple (dec_flags, next_input)
# where dec_flags is the second (integer) item of the decoder state
# and next_input is the chunk of input bytes that comes next after the
def line_buffering(self):
return self._line_buffering
+ @property
+ def write_through(self):
+ return self._write_through
+
@property
def buffer(self):
return self._buffer
+ def reconfigure(self, *, line_buffering=None, write_through=None):
+ """Reconfigure the text stream with new parameters.
+
+ This also flushes the stream.
+ """
+ if line_buffering is None:
+ line_buffering = self.line_buffering
+ if write_through is None:
+ write_through = self.write_through
+ self.flush()
+ self._configure(line_buffering, write_through)
+
def seekable(self):
if self.closed:
raise ValueError("I/O operation on closed file.")
self.assertEqual(t.encoding, "ascii")
self.assertEqual(t.errors, "strict")
self.assertFalse(t.line_buffering)
+ self.assertFalse(t.write_through)
def test_repr(self):
raw = self.BytesIO("hello".encode("utf-8"))
t.write("A\rB")
self.assertEqual(r.getvalue(), b"XY\nZA\rB")
+ def test_reconfigure_line_buffering(self):
+ r = self.BytesIO()
+ b = self.BufferedWriter(r, 1000)
+ t = self.TextIOWrapper(b, newline="\n", line_buffering=False)
+ t.write("AB\nC")
+ self.assertEqual(r.getvalue(), b"")
+
+ t.reconfigure(line_buffering=True) # implicit flush
+ self.assertEqual(r.getvalue(), b"AB\nC")
+ t.write("DEF\nG")
+ self.assertEqual(r.getvalue(), b"AB\nCDEF\nG")
+ t.write("H")
+ self.assertEqual(r.getvalue(), b"AB\nCDEF\nG")
+ t.reconfigure(line_buffering=False) # implicit flush
+ self.assertEqual(r.getvalue(), b"AB\nCDEF\nGH")
+ t.write("IJ")
+ self.assertEqual(r.getvalue(), b"AB\nCDEF\nGH")
+
+ # Keeping default value
+ t.reconfigure()
+ t.reconfigure(line_buffering=None)
+ self.assertEqual(t.line_buffering, False)
+ t.reconfigure(line_buffering=True)
+ t.reconfigure()
+ t.reconfigure(line_buffering=None)
+ self.assertEqual(t.line_buffering, True)
+
def test_default_encoding(self):
old_environ = dict(os.environ)
try:
self.assertTrue(write_called)
self.assertEqual(rawio.getvalue(), data * 11) # all flushed
+ def test_reconfigure_write_through(self):
+ raw = self.MockRawIO([])
+ t = self.TextIOWrapper(raw, encoding='ascii', newline='\n')
+ t.write('1')
+ t.reconfigure(write_through=True) # implied flush
+ self.assertEqual(t.write_through, True)
+ self.assertEqual(b''.join(raw._write_stack), b'1')
+ t.write('23')
+ self.assertEqual(b''.join(raw._write_stack), b'123')
+ t.reconfigure(write_through=False)
+ self.assertEqual(t.write_through, False)
+ t.write('45')
+ t.flush()
+ self.assertEqual(b''.join(raw._write_stack), b'12345')
+ # Keeping default value
+ t.reconfigure()
+ t.reconfigure(write_through=None)
+ self.assertEqual(t.write_through, False)
+ t.reconfigure(write_through=True)
+ t.reconfigure()
+ t.reconfigure(write_through=None)
+ self.assertEqual(t.write_through, True)
+
def test_read_nonbytes(self):
# Issue #17106
# Crash when underlying read() returns non-bytes
Library
-------
+- bpo-30526: Add TextIOWrapper.reconfigure() and a TextIOWrapper.write_through
+ attribute.
+
- bpo-30245: Fix possible overflow when organize struct.pack_into
error message. Patch by Yuan Liu.
return return_value;
}
+PyDoc_STRVAR(_io_TextIOWrapper_reconfigure__doc__,
+"reconfigure($self, /, *, line_buffering=None, write_through=None)\n"
+"--\n"
+"\n"
+"Reconfigure the text stream with new parameters.\n"
+"\n"
+"This also does an implicit stream flush.");
+
+#define _IO_TEXTIOWRAPPER_RECONFIGURE_METHODDEF \
+ {"reconfigure", (PyCFunction)_io_TextIOWrapper_reconfigure, METH_FASTCALL, _io_TextIOWrapper_reconfigure__doc__},
+
+static PyObject *
+_io_TextIOWrapper_reconfigure_impl(textio *self,
+ PyObject *line_buffering_obj,
+ PyObject *write_through_obj);
+
+static PyObject *
+_io_TextIOWrapper_reconfigure(textio *self, PyObject **args, Py_ssize_t nargs, PyObject *kwnames)
+{
+ PyObject *return_value = NULL;
+ static const char * const _keywords[] = {"line_buffering", "write_through", NULL};
+ static _PyArg_Parser _parser = {"|$OO:reconfigure", _keywords, 0};
+ PyObject *line_buffering_obj = Py_None;
+ PyObject *write_through_obj = Py_None;
+
+ if (!_PyArg_ParseStackAndKeywords(args, nargs, kwnames, &_parser,
+ &line_buffering_obj, &write_through_obj)) {
+ goto exit;
+ }
+ return_value = _io_TextIOWrapper_reconfigure_impl(self, line_buffering_obj, write_through_obj);
+
+exit:
+ return return_value;
+}
+
PyDoc_STRVAR(_io_TextIOWrapper_detach__doc__,
"detach($self, /)\n"
"--\n"
{
return _io_TextIOWrapper_close_impl(self);
}
-/*[clinic end generated code: output=8e5c21c88c7c70bc input=a9049054013a1b77]*/
+/*[clinic end generated code: output=7d0dc8eae4b725a1 input=a9049054013a1b77]*/
return -1;
}
+/* Return *default_value* if ob is None, 0 if ob is false, 1 if ob is true,
+ * -1 on error.
+ */
+static int
+convert_optional_bool(PyObject *obj, int default_value)
+{
+ long v;
+ if (obj == Py_None) {
+ v = default_value;
+ }
+ else {
+ v = PyLong_AsLong(obj);
+ if (v == -1 && PyErr_Occurred())
+ return -1;
+ }
+ return v != 0;
+}
+
+
+/*[clinic input]
+_io.TextIOWrapper.reconfigure
+ *
+ line_buffering as line_buffering_obj: object = None
+ write_through as write_through_obj: object = None
+
+Reconfigure the text stream with new parameters.
+
+This also does an implicit stream flush.
+
+[clinic start generated code]*/
+
+static PyObject *
+_io_TextIOWrapper_reconfigure_impl(textio *self,
+ PyObject *line_buffering_obj,
+ PyObject *write_through_obj)
+/*[clinic end generated code: output=7cdf79e7001e2856 input=baade27ecb9db7bc]*/
+{
+ int line_buffering;
+ int write_through;
+ PyObject *res;
+
+ line_buffering = convert_optional_bool(line_buffering_obj,
+ self->line_buffering);
+ write_through = convert_optional_bool(write_through_obj,
+ self->write_through);
+ if (line_buffering < 0 || write_through < 0) {
+ return NULL;
+ }
+ res = PyObject_CallMethodObjArgs((PyObject *) self, _PyIO_str_flush, NULL);
+ Py_XDECREF(res);
+ if (res == NULL) {
+ return NULL;
+ }
+ self->line_buffering = line_buffering;
+ self->write_through = write_through;
+ Py_RETURN_NONE;
+}
+
static int
textiowrapper_clear(textio *self)
{
static PyMethodDef textiowrapper_methods[] = {
_IO_TEXTIOWRAPPER_DETACH_METHODDEF
+ _IO_TEXTIOWRAPPER_RECONFIGURE_METHODDEF
_IO_TEXTIOWRAPPER_WRITE_METHODDEF
_IO_TEXTIOWRAPPER_READ_METHODDEF
_IO_TEXTIOWRAPPER_READLINE_METHODDEF
{"encoding", T_OBJECT, offsetof(textio, encoding), READONLY},
{"buffer", T_OBJECT, offsetof(textio, buffer), READONLY},
{"line_buffering", T_BOOL, offsetof(textio, line_buffering), READONLY},
+ {"write_through", T_BOOL, offsetof(textio, write_through), READONLY},
{"_finalizing", T_BOOL, offsetof(textio, finalizing), 0},
{NULL}
};
projects / python / commitdiff