Skip to content

Make BufferedWriter generic over a protocol #14015

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 5 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion stdlib/@tests/test_cases/check_io.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,17 @@
from _io import BufferedReader
from _io import BufferedReader, BufferedRWPair, BufferedWriter
from gzip import GzipFile
from io import FileIO, RawIOBase, TextIOWrapper
from socket import SocketIO
from typing import Any
from typing_extensions import assert_type

socket: Any = None

BufferedReader(RawIOBase())
BufferedWriter(RawIOBase())
BufferedWriter(SocketIO(socket, "r"))

BufferedRWPair(open("", "rb"), open("", "wb"))

assert_type(TextIOWrapper(FileIO("")).buffer, FileIO)
assert_type(TextIOWrapper(FileIO(13)).detach(), FileIO)
Expand Down
37 changes: 32 additions & 5 deletions stdlib/_io.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,7 @@ 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: ...

@type_check_only
class _BufferedReaderStream(Protocol):
def read(self, n: int = ..., /) -> bytes: ...
# Optional: def readall(self) -> bytes: ...
Expand Down Expand Up @@ -122,9 +123,35 @@ class BufferedReader(BufferedIOBase, _BufferedIOBase, BinaryIO, Generic[_Buffere
def seek(self, target: int, whence: int = 0, /) -> int: ...
def truncate(self, pos: int | None = None, /) -> int: ...

class BufferedWriter(BufferedIOBase, _BufferedIOBase, BinaryIO): # type: ignore[misc] # incompatible definitions of writelines in the base classes
raw: RawIOBase
def __init__(self, raw: RawIOBase, buffer_size: int = 8192) -> None: ...
@type_check_only
class _BufferedWriterStream(Protocol):
def write(self, b: WriteableBuffer, /) -> 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 writable(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: ...

_BufferedWriterStreamT = TypeVar("_BufferedWriterStreamT", bound=_BufferedWriterStream, default=_BufferedWriterStream)

class BufferedWriter(BufferedIOBase, _BufferedIOBase, BinaryIO, Generic[_BufferedWriterStreamT]): # type: ignore[misc] # incompatible definitions of writelines in the base classes
raw: _BufferedWriterStreamT
def __init__(self, raw: _BufferedWriterStreamT, buffer_size: int = 8192) -> None: ...
def write(self, buffer: ReadableBuffer, /) -> int: ...
def seek(self, target: int, whence: int = 0, /) -> int: ...
def truncate(self, pos: int | None = None, /) -> int: ...
Expand All @@ -138,8 +165,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, Generic[_BufferedReaderStreamT]):
def __init__(self, reader: _BufferedReaderStreamT, writer: RawIOBase, buffer_size: int = 8192, /) -> None: ...
class BufferedRWPair(BufferedIOBase, _BufferedIOBase, Generic[_BufferedReaderStreamT, _BufferedWriterStreamT]):
def __init__(self, reader: _BufferedReaderStreamT, writer: _BufferedWriterStreamT, buffer_size: int = 8192, /) -> None: ...
def peek(self, size: int = 0, /) -> bytes: ...

class _TextIOBase(_IOBase):
Expand Down