Add Reader and Writer protocols

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Unreleased
 
+New features:
+
+- Add `typing_extensions.Reader` and `typing_extensions.Writer`. Patch by
+  Sebastian Rittau.
+
+Bugfixes:
+
 - Drop support for Python 3.8 (including PyPy-3.8). Patch by [Victorien Plot](https://github.com/Viicos).
 - Add support for inline typed dictionaries ([PEP 764](https://peps.python.org/pep-0764/)).
   Patch by [Victorien Plot](https://github.com/Viicos).
@@ -17,6 +24,7 @@
 # Release 4.13.1 (April 3, 2025)
 
 Bugfixes:
+
 - Fix regression in 4.13.0 on Python 3.10.2 causing a `TypeError` when using `Concatenate`.
   Patch by [Daraan](https://github.com/Daraan).
 - Fix `TypeError` when using `evaluate_forward_ref` on Python 3.10.1-2 and 3.9.8-10.

src/test_typing_extensions.py

@@ -4103,6 +4103,32 @@ def foo(self): pass
         self.assertIsSubclass(Bar, Functor)
 
 
+class SpecificProtocolTests(BaseTestCase):
+    def test_reader_runtime_checkable(self):
+        class MyReader:
+            def read(self, n: int) -> bytes:
+                return b""
+
+        class WrongReader:
+            def readx(self, n: int) -> bytes:
+                return b""
+
+        assert isinstance(MyReader(), typing_extensions.Reader)
+        assert not isinstance(WrongReader(), typing_extensions.Reader)
+
+    def test_writer_runtime_checkable(self):
+        class MyWriter:
+            def write(self, b: bytes) -> int:
+                return 0
+
+        class WrongWriter:
+            def writex(self, b: bytes) -> int:
+                return 0
+
+        assert isinstance(MyWriter(), typing_extensions.Writer)
+        assert not isinstance(WrongWriter(), typing_extensions.Writer)
+
+
 class Point2DGeneric(Generic[T], TypedDict):
     a: T
     b: T

src/typing_extensions.py

@@ -6,6 +6,7 @@
 import enum
 import functools
 import inspect
+import io
 import keyword
 import operator
 import sys
@@ -56,6 +57,8 @@
     'SupportsIndex',
     'SupportsInt',
     'SupportsRound',
+    'Reader',
+    'Writer',
 
     # One-off things.
     'Annotated',
@@ -846,6 +849,56 @@ def __round__(self, ndigits: int = 0) -> T_co:
             pass
 
 
+if hasattr(io, "Reader") and hasattr(io, "Writer"):
+    Reader = io.Reader
+    Writer = io.Writer
+else:
+    @runtime_checkable
+    class Reader(Protocol[T_co]):
+        """Protocol for simple I/O reader instances.
+
+        This protocol only supports blocking I/O.
+        """
+
+        __slots__ = ()
+
+        @abc.abstractmethod
+        def read(self, size: int = ..., /) -> T_co:
+            """Read data from the input stream and return it.
+
+            If size is specified, at most size items (bytes/characters) will be
+            read.
+            """
+
+    @runtime_checkable
+    class Writer(Protocol[T_contra]):
+        """Protocol for simple I/O writer instances.
+
+        This protocol only supports blocking I/O.
+        """
+
+        __slots__ = ()
+
+        @abc.abstractmethod
+        def write(self, data: T_contra, /) -> int:
+            """Write data to the output stream and return the number of items written."""
+
+
+def _ensure_subclassable(mro_entries):
+    def inner(func):
+        if sys.implementation.name == "pypy" and sys.version_info < (3, 9):
+            cls_dict = {
+                "__call__": staticmethod(func),
+                "__mro_entries__": staticmethod(mro_entries)
+            }
+            t = type(func.__name__, (), cls_dict)
+            return functools.update_wrapper(t(), func)
+        else:
+            func.__mro_entries__ = mro_entries
+            return func
+    return inner
+
+
 _NEEDS_SINGLETONMETA = (
     not hasattr(typing, "NoDefault") or not hasattr(typing, "NoExtraItems")
 )