gh-127647: Add typing.Reader and Writer protocols (#127648)

1 files changed, 55 insertions, 1 deletions

diff --git a/Lib/io.py b/Lib/io.py
index f0e2fa1..e9fe619 100644
--- a/Lib/io.py
+++ b/Lib/io.py
@@ -46,12 +46,14 @@ __all__ = ["BlockingIOError", "open", "open_code", "IOBase", "RawIOBase",
            "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 @@ except ImportError:
     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)