gh-127647: Add typing.Reader and Writer protocols

Doc/library/io.rst

@@ -1147,6 +1147,49 @@ 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 :func:`@typing.runtime_checkable <typing.runtime_checkable>`.
+
+.. class:: Reader[T]
+
+   Protocol for reading from a file or other input stream.
+
+   .. versionadded:: next
+
+   .. method:: read(size=..., /)
+
+      Read data from the input stream and return it. If ``size`` is
+      specified, 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]
+
+   Protocol for writing to a file or other output stream.
+
+   .. versionadded:: next
+
+   .. method:: write(data, /)
+
+      Write data to the output stream and return 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 used for
+static type checking.
+
 Performance
 -----------
 

Doc/library/typing.rst

@@ -2803,17 +2803,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

@@ -602,6 +602,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 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/typing.py

@@ -1948,6 +1948,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`.