diff --git a/Lib/_pyio.py b/Lib/_pyio.py
--- a/Lib/_pyio.py
+++ b/Lib/_pyio.py
@@ -1472,7 +1472,7 @@ class TextIOWrapper(TextIOBase):
     _CHUNK_SIZE = 2048
 
     def __init__(self, buffer, encoding=None, errors=None, newline=None,
-                 line_buffering=False):
+                 line_buffering=False, write_through=False):
         if newline is not None and not isinstance(newline, str):
             raise TypeError("illegal newline type: %r" % (type(newline),))
         if newline not in (None, "", "\n", "\r", "\r\n"):
@@ -1515,6 +1515,7 @@ class TextIOWrapper(TextIOBase):
         self._decoded_chars_used = 0  # offset into _decoded_chars for read()
         self._snapshot = None  # info for reconstructing decoder state
         self._seekable = self._telling = self.buffer.seekable()
+        self._has_read1 = hasattr(self.buffer, 'read1')
 
         if self._seekable and self.writable():
             position = self.buffer.tell()
@@ -1680,7 +1681,10 @@ class TextIOWrapper(TextIOBase):
             # len(dec_buffer) bytes ago with decoder state (b'', dec_flags).
 
         # Read a chunk, decode it, and put the result in self._decoded_chars.
-        input_chunk = self.buffer.read1(self._CHUNK_SIZE)
+        if self._has_read1:
+            input_chunk = self.buffer.read1(self._CHUNK_SIZE)
+        else:
+            input_chunk = self.buffer.read(self._CHUNK_SIZE)
         eof = not input_chunk
         self._set_decoded_chars(self._decoder.decode(input_chunk, eof))
 
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -2307,6 +2307,26 @@ class TextIOWrapperTest(unittest.TestCas
         with self.assertRaises(AttributeError):
             txt.buffer = buf
 
+    def test_rawio(self):
+        # Issue #12591: TextIOWrapper must work with raw I/O objects, so
+        # that subprocess.Popen() can have the required unbuffered
+        # semantics with universal_newlines=True.
+        raw = self.MockRawIO([b'abc', b'def', b'ghi\njkl\nopq\n'])
+        txt = self.TextIOWrapper(raw, encoding='ascii')
+        # Reads
+        self.assertEqual(txt.read(4), 'abcd')
+        self.assertEqual(txt.readline(), 'efghi\n')
+        self.assertEqual(list(txt), ['jkl\n', 'opq\n'])
+
+    def test_rawio_write_through(self):
+        # Issue #12591: with write_through=True, writes don't need a flush
+        raw = self.MockRawIO([b'abc', b'def', b'ghi\njkl\nopq\n'])
+        txt = self.TextIOWrapper(raw, encoding='ascii', write_through=True)
+        txt.write('1')
+        txt.write('23\n4')
+        txt.write('5')
+        self.assertEqual(b''.join(raw._write_stack), b'123\n45')
+
 class CTextIOWrapperTest(TextIOWrapperTest):
 
     def test_initialization(self):
diff --git a/Modules/_io/textio.c b/Modules/_io/textio.c
--- a/Modules/_io/textio.c
+++ b/Modules/_io/textio.c
@@ -653,10 +653,12 @@ typedef struct
     PyObject *errors;
     const char *writenl; /* utf-8 encoded, NULL stands for \n */
     char line_buffering;
+    char write_through;
     char readuniversal;
     char readtranslate;
     char writetranslate;
     char seekable;
+    char has_read1;
     char telling;
     char deallocating;
     /* Specialized encoding func (see below) */
@@ -809,13 +811,13 @@ static int
 textiowrapper_init(textio *self, PyObject *args, PyObject *kwds)
 {
     char *kwlist[] = {"buffer", "encoding", "errors",
-                      "newline", "line_buffering",
+                      "newline", "line_buffering", "write_through",
                       NULL};
     PyObject *buffer, *raw;
     char *encoding = NULL;
     char *errors = NULL;
     char *newline = NULL;
-    int line_buffering = 0;
+    int line_buffering = 0, write_through = 0;
     _PyIO_State *state = IO_STATE;
 
     PyObject *res;
@@ -823,9 +825,9 @@ textiowrapper_init(textio *self, PyObjec
 
     self->ok = 0;
     self->detached = 0;
-    if (!PyArg_ParseTupleAndKeywords(args, kwds, "O|zzzi:fileio",
+    if (!PyArg_ParseTupleAndKeywords(args, kwds, "O|zzzii:fileio",
                                      kwlist, &buffer, &encoding, &errors,
-                                     &newline, &line_buffering))
+                                     &newline, &line_buffering, &write_through))
         return -1;
 
     if (newline && newline[0] != '\0'
@@ -930,6 +932,7 @@ textiowrapper_init(textio *self, PyObjec
     self->chunk_size = 8192;
     self->readuniversal = (newline == NULL || newline[0] == '\0');
     self->line_buffering = line_buffering;
+    self->write_through = write_through;
     self->readtranslate = (newline == NULL);
     if (newline) {
         self->readnl = PyUnicode_FromString(newline);
@@ -1039,6 +1042,8 @@ textiowrapper_init(textio *self, PyObjec
     self->seekable = self->telling = PyObject_IsTrue(res);
     Py_DECREF(res);
 
+    self->has_read1 = PyObject_HasAttrString(buffer, "read1");
+
     self->encoding_start_of_stream = 0;
     if (self->seekable && self->encoder) {
         PyObject *cookieObj;
@@ -1282,7 +1287,9 @@ textiowrapper_write(textio *self, PyObje
         text = newtext;
     }
 
-    if (self->line_buffering &&
+    if (self->write_through)
+        needflush = 1;
+    else if (self->line_buffering &&
         (haslf ||
          findchar(PyUnicode_AS_UNICODE(text),
                   PyUnicode_GET_SIZE(text), '\r')))
@@ -1429,7 +1436,8 @@ textiowrapper_read_chunk(textio *self)
     if (chunk_size == NULL)
         goto fail;
     input_chunk = PyObject_CallMethodObjArgs(self->buffer,
-        _PyIO_str_read1, chunk_size, NULL);
+        (self->has_read1 ? _PyIO_str_read1: _PyIO_str_read),
+        chunk_size, NULL);
     Py_DECREF(chunk_size);
     if (input_chunk == NULL)
         goto fail;