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
-----------
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
------------------------
: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
----
_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') :
"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,
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)
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,
from functools import lru_cache, wraps, reduce
import gc
import inspect
+import io
import itertools
import operator
import os
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):
'Reversible', 'Buffer',
],
'contextlib': ['AbstractContextManager', 'AbstractAsyncContextManager'],
+ 'io': ['Reader', 'Writer'],
'os': ['PathLike'],
}
--- /dev/null
+Add protocols :class:`io.Reader` and :class:`io.Writer` as
+alternatives to :class:`typing.IO`, :class:`typing.TextIO`, and
+:class:`typing.BinaryIO`.
projects / thirdparty / Python / cpython.git / commitdiff
