bpo-41305: Add StreamReader.readinto()

Doc/library/asyncio-stream.rst

@@ -211,6 +211,13 @@ StreamReader
 
       .. versionadded:: 3.5.2
 
+   .. coroutinemethod:: readinto(buf)
+
+      Read up to *n* bytes with *n* being equal to the length of *buf* and
+      copy the buffer read from the stream into *buf*.
+
+      Return the number of bytes read from the stream.
+
    .. method:: at_eof()
 
       Return ``True`` if the buffer is empty and :meth:`feed_eof`

Lib/asyncio/streams.py

@@ -690,6 +690,44 @@ async def read(self, n=-1):
         self._maybe_resume_transport()
         return data
 
+    async def readinto(self, buf):
+        """Read up to the length of `buf` number of bytes from the stream and
+        fill the `buf` bytes object with what was read from the stream.
+
+        Returns the number of bytes read from the stream.
+
+        If `buf` is empty, return 0 immediately.
+
+        If `buf` is not empty, this function tries to read the length of `buf`
+        number of bytes, and may fill the buffer with less or equal bytes than
+        requested, but at least one byte. If EOF was received before any byte
+        is read, this function returns 0.
+
+        Returned value is not limited with limit, configured at stream
+        creation.
+
+        If stream was paused, this function will automatically resume it if
+        needed.
+        """
+        n = len(buf)
+
+        if self._exception is not None:
+            raise self._exception
+
+        if n == 0:
+            return 0
+
+        if not self._buffer and not self._eof:
+            await self._wait_for_data('readinto')
+
+        length = min(n, len(self._buffer))
+        if length > 0:
+            buf[:length] = self._buffer[:length]
+
+        del self._buffer[:n]
+        self._maybe_resume_transport()
+        return length
+
     async def readexactly(self, n):
         """Read exactly `n` bytes.
 

Lib/test/test_asyncio/test_streams.py

@@ -236,6 +236,80 @@ def test_read_limit(self):
         self.assertEqual(b'chunk', data)
         self.assertEqual(b'', stream._buffer)
 
+    def test_readinto_empty_buffer(self):
+        # Read zero bytes.
+        stream = asyncio.StreamReader(loop=self.loop)
+        stream.feed_data(self.DATA)
+
+        data = bytearray(0)
+        length = self.loop.run_until_complete(stream.readinto(data))
+        self.assertEqual(0, length)
+        self.assertEqual(b'', data)
+        self.assertEqual(self.DATA, stream._buffer)
+
+    def test_readinto(self):
+        # Read bytes.
+        stream = asyncio.StreamReader(loop=self.loop)
+        data = bytearray(30)
+        read_task = self.loop.create_task(stream.readinto(data))
+
+        def cb():
+            stream.feed_data(self.DATA)
+        self.loop.call_soon(cb)
+
+        length = self.loop.run_until_complete(read_task)
+        self.assertEqual(len(self.DATA), length)
+        self.assertEqual(self.DATA, data[:length])
+        self.assertEqual(b'', stream._buffer)
+
+    def test_readinto_line_breaks(self):
+        # Read bytes without line breaks.
+        stream = asyncio.StreamReader(loop=self.loop)
+        stream.feed_data(b'line1')
+        stream.feed_data(b'line2')
+
+        data = bytearray(5)
+        length = self.loop.run_until_complete(stream.readinto(data))
+
+        self.assertEqual(5, length)
+        self.assertEqual(b'line1', data)
+        self.assertEqual(b'line2', stream._buffer)
+
+    def test_readinto_eof(self):
+        # Read bytes, stop at eof.
+        stream = asyncio.StreamReader(loop=self.loop)
+        data = bytearray(1024)
+        read_task = self.loop.create_task(stream.readinto(data))
+
+        def cb():
+            stream.feed_eof()
+        self.loop.call_soon(cb)
+
+        length = self.loop.run_until_complete(read_task)
+        self.assertEqual(0, length)
+        self.assertEqual(b'', stream._buffer)
+
+    def test_readinto_exception(self):
+        stream = asyncio.StreamReader(loop=self.loop)
+        stream.feed_data(b'line\n')
+
+        data = bytearray(2)
+        length = self.loop.run_until_complete(stream.readinto(data))
+        self.assertEqual(b'li', data)
+
+        stream.set_exception(ValueError())
+        self.assertRaises(
+            ValueError, self.loop.run_until_complete, stream.readinto(data))
+
+    def test_readinto_limit(self):
+        stream = asyncio.StreamReader(limit=3, loop=self.loop)
+        stream.feed_data(b'chunk')
+        data = bytearray(5)
+        length = self.loop.run_until_complete(stream.readinto(data))
+        self.assertEqual(5, length)
+        self.assertEqual(b'chunk', data)
+        self.assertEqual(b'', stream._buffer)
+
     def test_readline(self):
         # Read one line. 'readline' will need to wait for the data
         # to come from 'cb'

Misc/NEWS.d/next/Library/2020-07-15-19-34-39.bpo-41305.3In2Tp.rst

@@ -0,0 +1 @@
+Add the ``StreamReader.readinto(buf)`` function.