Add Reader and Writer protocols

CHANGELOG.md

@@ -1,12 +1,20 @@
 # Unreleased
 
+New features:
+
+- Add `typing_extensions.Reader` and `typing_extensions.Writer`. Patch by
+  Sebastian Rittau.
+
+Bugfixes:
+
 - Fix `TypeError` when taking the union of `typing_extensions.TypeAliasType` and a
   `typing.TypeAliasType` on Python 3.12 and 3.13.
   Patch by [Joren Hammudoglu](https://github.com/jorenham).
 
 # 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

@@ -4088,6 +4088,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',
@@ -863,6 +866,41 @@ 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):