Make BufferedReader generic over a protocol (#13533)

This commit is contained in:
Sebastian Rittau
2025-05-11 16:24:31 +02:00
committed by GitHub
parent b444909873
commit 1ebbbeec75
2 changed files with 37 additions and 8 deletions
+4 -1
View File
@@ -1,7 +1,10 @@
from _io import BufferedReader
from gzip import GzipFile
from io import FileIO, TextIOWrapper
from io import FileIO, RawIOBase, TextIOWrapper
from typing_extensions import assert_type
BufferedReader(RawIOBase())
assert_type(TextIOWrapper(FileIO("")).buffer, FileIO)
assert_type(TextIOWrapper(FileIO(13)).detach(), FileIO)
assert_type(TextIOWrapper(GzipFile("")).buffer, GzipFile)
+33 -7
View File
@@ -88,9 +88,36 @@ class BytesIO(BufferedIOBase, _BufferedIOBase, BinaryIO): # type: ignore[misc]
def readlines(self, size: int | None = None, /) -> list[bytes]: ...
def seek(self, pos: int, whence: int = 0, /) -> int: ...
class BufferedReader(BufferedIOBase, _BufferedIOBase, BinaryIO): # type: ignore[misc] # incompatible definitions of methods in the base classes
raw: RawIOBase
def __init__(self, raw: RawIOBase, buffer_size: int = 8192) -> None: ...
class _BufferedReaderStream(Protocol):
def read(self, n: int = ..., /) -> bytes: ...
# Optional: def readall(self) -> bytes: ...
def readinto(self, b: memoryview, /) -> int | None: ...
def seek(self, pos: int, whence: int, /) -> int: ...
def tell(self) -> int: ...
def truncate(self, size: int, /) -> int: ...
def flush(self) -> object: ...
def close(self) -> object: ...
@property
def closed(self) -> bool: ...
def readable(self) -> bool: ...
def seekable(self) -> bool: ...
# The following methods just pass through to the underlying stream. Since
# not all streams support them, they are marked as optional here, and will
# raise an AttributeError if called on a stream that does not support them.
# @property
# def name(self) -> Any: ... # Type is inconsistent between the various I/O types.
# @property
# def mode(self) -> str: ...
# def fileno(self) -> int: ...
# def isatty(self) -> bool: ...
_BufferedReaderStreamT = TypeVar("_BufferedReaderStreamT", bound=_BufferedReaderStream, default=_BufferedReaderStream)
class BufferedReader(BufferedIOBase, _BufferedIOBase, BinaryIO, Generic[_BufferedReaderStreamT]): # type: ignore[misc] # incompatible definitions of methods in the base classes
raw: _BufferedReaderStreamT
def __init__(self, raw: _BufferedReaderStreamT, buffer_size: int = 8192) -> None: ...
def peek(self, size: int = 0, /) -> bytes: ...
def seek(self, target: int, whence: int = 0, /) -> int: ...
def truncate(self, pos: int | None = None, /) -> int: ...
@@ -111,8 +138,8 @@ class BufferedRandom(BufferedIOBase, _BufferedIOBase, BinaryIO): # type: ignore
def peek(self, size: int = 0, /) -> bytes: ...
def truncate(self, pos: int | None = None, /) -> int: ...
class BufferedRWPair(BufferedIOBase, _BufferedIOBase):
def __init__(self, reader: RawIOBase, writer: RawIOBase, buffer_size: int = 8192, /) -> None: ...
class BufferedRWPair(BufferedIOBase, _BufferedIOBase, Generic[_BufferedReaderStreamT]):
def __init__(self, reader: _BufferedReaderStreamT, writer: RawIOBase, buffer_size: int = 8192, /) -> None: ...
def peek(self, size: int = 0, /) -> bytes: ...
class _TextIOBase(_IOBase):
@@ -131,8 +158,7 @@ class _TextIOBase(_IOBase):
@type_check_only
class _WrappedBuffer(Protocol):
# "name" is wrapped by TextIOWrapper. Its type is inconsistent between
# the various I/O types, see the comments on TextIOWrapper.name and
# TextIO.name.
# the various I/O types.
@property
def name(self) -> Any: ...
@property