gh-127647: Add typing.Reader and Writer protocols

Doc/library/io.rst

@@ -1147,6 +1147,55 @@ Text I/O
    It inherits from :class:`codecs.IncrementalDecoder`.
 
 
+Static Typing
+-------------
+
+The following protocols can be used for annotating function and method
+arguments for simple stream reading or writing operations. They are decorated
+with :deco:`typing.runtime_checkable`.
+
+.. class:: Reader[T]
+
+   Generic protocol for reading from a file or other input stream. ``T`` will
+   usually be :class:`str` or :class:`bytes`, but can be any type that is
+   read from the stream.
+
+   .. versionadded:: next
+
+   .. method:: read()
+               read(size, /)
+
+      Read data from the input stream and return it. If *size* is
+      specified, it should be an integer, and at most *size* items
+      (bytes/characters) will be read.
+
+   For example::
+
+     def read_it(reader: Reader[str]):
+         data = reader.read(11)
+         assert isinstance(data, str)
+
+.. class:: Writer[T]
+
+   Generic protocol for writing to a file or other output stream. ``T`` will
+   usually be :class:`str` or :class:`bytes`, but can be any type that can be
+   written to the stream.
+
+   .. versionadded:: next
+
+   .. method:: write(data, /)
+
+      Write *data* to the output stream and return the number of items
+      (bytes/characters) written.
+
+   For example::
+
+     def write_binary(writer: Writer[bytes]):
+         writer.write(b"Hello world!\n")
+
+See :ref:`typing-io` for other I/O related protocols and classes that can be
+used for static type checking.
+
 Performance
 -----------
 

Doc/library/typing.rst

@@ -2834,17 +2834,35 @@ with :func:`@runtime_checkable <runtime_checkable>`.
     An ABC with one abstract method ``__round__``
     that is covariant in its return type.
 
-ABCs for working with IO
-------------------------
+.. _typing-io:
+
+ABCs and Protocols for working with I/O
+---------------------------------------
 
-.. class:: IO
-           TextIO
-           BinaryIO
+.. class:: IO[AnyStr]
+           TextIO[AnyStr]
+           BinaryIO[AnyStr]
 
-   Generic type ``IO[AnyStr]`` and its subclasses ``TextIO(IO[str])``
+   Generic class ``IO[AnyStr]`` and its subclasses ``TextIO(IO[str])``
    and ``BinaryIO(IO[bytes])``
    represent the types of I/O streams such as returned by
-   :func:`open`.
+   :func:`open`. Please note that these classes are not protocols, and
+   their interface is fairly broad.
+
+The protocols :class:`io.Reader` and :class:`io.Writer` offer a simpler
+alternative for argument types, when only the ``read()`` or ``write()``
+methods are accessed, respectively::
+
+   def read_and_write(reader: Reader[str], writer: Writer[bytes]):
+       data = reader.read()
+       writer.write(data.encode())
+
+Also consider using :class:`collections.abc.Iterable` for iterating over
+the lines of an input stream::
+
+   def read_config(stream: Iterable[str]):
+       for line in stream:
+           ...
 
 Functions and decorators
 ------------------------

Doc/whatsnew/3.14.rst

@@ -619,6 +619,11 @@ io
   :exc:`BlockingIOError` if the operation cannot immediately return bytes.
   (Contributed by Giovanni Siragusa in :gh:`109523`.)
 
+* Add protocols :class:`io.Reader` and :class:`io.Writer` as a simpler
+  alternatives to the pseudo-protocols :class:`typing.IO`,
+  :class:`typing.TextIO`, and :class:`typing.BinaryIO`.
+  (Contributed by Sebastian Rittau in :gh:`127648`.)
+
 
 json
 ----

Lib/_pyio.py

@@ -16,7 +16,7 @@
     _setmode = None
 
 import io
-from io import (__all__, SEEK_SET, SEEK_CUR, SEEK_END)  # noqa: F401
+from io import (__all__, SEEK_SET, SEEK_CUR, SEEK_END, Reader, Writer)  # noqa: F401
 
 valid_seek_flags = {0, 1, 2}  # Hardwired values
 if hasattr(os, 'SEEK_HOLE') :

Lib/io.py

@@ -46,12 +46,14 @@
            "BufferedReader", "BufferedWriter", "BufferedRWPair",
            "BufferedRandom", "TextIOBase", "TextIOWrapper",
            "UnsupportedOperation", "SEEK_SET", "SEEK_CUR", "SEEK_END",
-           "DEFAULT_BUFFER_SIZE", "text_encoding", "IncrementalNewlineDecoder"]
+           "DEFAULT_BUFFER_SIZE", "text_encoding", "IncrementalNewlineDecoder",
+           "Reader", "Writer"]
 
 
 import _io
 import abc
 
+from _collections_abc import _check_methods
 from _io import (DEFAULT_BUFFER_SIZE, BlockingIOError, UnsupportedOperation,
                  open, open_code, FileIO, BytesIO, StringIO, BufferedReader,
                  BufferedWriter, BufferedRWPair, BufferedRandom,
@@ -97,3 +99,55 @@ class TextIOBase(_io._TextIOBase, IOBase):
     pass
 else:
     RawIOBase.register(_WindowsConsoleIO)
+
+#
+# Static Typing Support
+#
+
+GenericAlias = type(list[int])
+
+
+class Reader(metaclass=abc.ABCMeta):
+    """Protocol for simple I/O reader instances.
+
+    This protocol only supports blocking I/O.
+    """
+
+    __slots__ = ()
+
+    @abc.abstractmethod
+    def read(self, size=..., /):
+        """Read data from the input stream and return it.
+
+        If *size* is specified, at most *size* items (bytes/characters) will be
+        read.
+        """
+
+    @classmethod
+    def __subclasshook__(cls, C):
+        if cls is Reader:
+            return _check_methods(C, "read")
+        return NotImplemented
+
+    __class_getitem__ = classmethod(GenericAlias)
+
+
+class Writer(metaclass=abc.ABCMeta):
+    """Protocol for simple I/O writer instances.
+
+    This protocol only supports blocking I/O.
+    """
+
+    __slots__ = ()
+
+    @abc.abstractmethod
+    def write(self, data, /):
+        """Write *data* to the output stream and return the number of items written."""
+
+    @classmethod
+    def __subclasshook__(cls, C):
+        if cls is Writer:
+            return _check_methods(C, "write")
+        return NotImplemented
+
+    __class_getitem__ = classmethod(GenericAlias)

Lib/test/test_io.py

@@ -4916,6 +4916,24 @@ class PySignalsTest(SignalsTest):
     test_reentrant_write_text = None
 
 
+class ProtocolsTest(unittest.TestCase):
+    class MyReader:
+        def read(self, sz=-1):
+            return b""
+
+    class MyWriter:
+        def write(self, b: bytes):
+            pass
+
+    def test_reader_subclass(self):
+        self.assertIsSubclass(MyReader, io.Reader[bytes])
+        self.assertNotIsSubclass(str, io.Reader[bytes])
+
+    def test_writer_subclass(self):
+        self.assertIsSubclass(MyWriter, io.Writer[bytes])
+        self.assertNotIsSubclass(str, io.Writer[bytes])
+
+
 def load_tests(loader, tests, pattern):
     tests = (CIOTest, PyIOTest, APIMismatchTest,
              CBufferedReaderTest, PyBufferedReaderTest,

Lib/test/test_typing.py

@@ -6,6 +6,7 @@
 from functools import lru_cache, wraps, reduce
 import gc
 import inspect
+import io
 import itertools
 import operator
 import os
@@ -4294,6 +4295,40 @@ def __release_buffer__(self, mv: memoryview) -> None:
         self.assertNotIsSubclass(C, ReleasableBuffer)
         self.assertNotIsInstance(C(), ReleasableBuffer)
 
+    def test_io_reader_protocol_allowed(self):
+        @runtime_checkable
+        class CustomReader(io.Reader[bytes], Protocol):
+            def close(self): ...
+
+        class A: pass
+        class B:
+            def read(self, sz=-1):
+                return b""
+            def close(self):
+                pass
+
+        self.assertIsSubclass(B, CustomReader)
+        self.assertIsInstance(B(), CustomReader)
+        self.assertNotIsSubclass(A, CustomReader)
+        self.assertNotIsInstance(A(), CustomReader)
+
+    def test_io_writer_protocol_allowed(self):
+        @runtime_checkable
+        class CustomWriter(io.Writer[bytes], Protocol):
+            def close(self): ...
+
+        class A: pass
+        class B:
+            def write(self, b):
+                pass
+            def close(self):
+                pass
+
+        self.assertIsSubclass(B, CustomWriter)
+        self.assertIsInstance(B(), CustomWriter)
+        self.assertNotIsSubclass(A, CustomWriter)
+        self.assertNotIsInstance(A(), CustomWriter)
+
     def test_builtin_protocol_allowlist(self):
         with self.assertRaises(TypeError):
             class CustomProtocol(TestCase, Protocol):

Lib/typing.py

@@ -1876,6 +1876,7 @@ def _allow_reckless_class_checks(depth=2):
         'Reversible', 'Buffer',
     ],
     'contextlib': ['AbstractContextManager', 'AbstractAsyncContextManager'],
+    'io': ['Reader', 'Writer'],
     'os': ['PathLike'],
 }
 

Misc/NEWS.d/next/Library/2024-12-05-19-54-16.gh-issue-127647.Xd78Vs.rst

@@ -0,0 +1,3 @@
+Add protocols :class:`io.Reader` and :class:`io.Writer` as
+alternatives to :class:`typing.IO`, :class:`typing.TextIO`, and
+:class:`typing.BinaryIO`.