Reuse more IO-related Protocols

stdlib/bz2.pyi

@@ -1,23 +1,26 @@
 import _compression
 import sys
 from _compression import BaseStream
-from _typeshed import ReadableBuffer, StrOrBytesPath, WriteableBuffer
+from _typeshed import ReadableBuffer, StrOrBytesPath, SupportsWrite, WriteableBuffer
 from collections.abc import Iterable
-from typing import IO, Any, Literal, Protocol, SupportsIndex, TextIO, final, overload
+from typing import IO, Any, Literal, Protocol, SupportsIndex, TextIO, final, overload, type_check_only
 from typing_extensions import Self, TypeAlias
 
 __all__ = ["BZ2File", "BZ2Compressor", "BZ2Decompressor", "open", "compress", "decompress"]
 
-# The following attributes and methods are optional:
-# def fileno(self) -> int: ...
-# def close(self) -> object: ...
-class _ReadableFileobj(_compression._Reader, Protocol): ...
+@type_check_only
+class _ReadableFileobj(_compression._Reader, Protocol):
+    # The following attributes and methods are optional:
+    # def fileno(self) -> int: ...
+    # def close(self) -> object: ...
+    ...
 
-class _WritableFileobj(Protocol):
-    def write(self, b: bytes, /) -> object: ...
+@type_check_only
+class _WritableFileobj(SupportsWrite[bytes], Protocol):
     # The following attributes and methods are optional:
     # def fileno(self) -> int: ...
     # def close(self) -> object: ...
+    ...
 
 def compress(data: ReadableBuffer, compresslevel: int = 9) -> bytes: ...
 def decompress(data: ReadableBuffer) -> bytes: ...

stdlib/codecs.pyi

@@ -1,9 +1,9 @@
 import types
 from _codecs import *
-from _typeshed import ReadableBuffer
+from _typeshed import ReadableBuffer, SupportsRead, SupportsWrite
 from abc import abstractmethod
 from collections.abc import Callable, Generator, Iterable
-from typing import Any, BinaryIO, Final, Literal, Protocol, TextIO
+from typing import Any, BinaryIO, Final, Literal, Protocol, TextIO, type_check_only
 from typing_extensions import Self
 
 __all__ = [
@@ -58,16 +58,17 @@ BOM32_LE: Final = b"\xff\xfe"
 BOM64_BE: Final = b"\x00\x00\xfe\xff"
 BOM64_LE: Final = b"\xff\xfe\x00\x00"
 
-class _WritableStream(Protocol):
-    def write(self, data: bytes, /) -> object: ...
+@type_check_only
+class _WritableStream(SupportsWrite[bytes], Protocol):
     def seek(self, offset: int, whence: int, /) -> object: ...
     def close(self) -> object: ...
 
-class _ReadableStream(Protocol):
-    def read(self, size: int = ..., /) -> bytes: ...
+@type_check_only
+class _ReadableStream(SupportsRead[bytes], Protocol):
     def seek(self, offset: int, whence: int, /) -> object: ...
     def close(self) -> object: ...
 
+@type_check_only
 class _Stream(_WritableStream, _ReadableStream, Protocol): ...
 
 # TODO: this only satisfies the most common interface, where

stdlib/fileinput.pyi

@@ -1,8 +1,8 @@
 import sys
-from _typeshed import AnyStr_co, StrOrBytesPath
+from _typeshed import AnyStr_co, HasFileno, StrOrBytesPath, SupportsNoArgReadline
 from collections.abc import Callable, Iterable, Iterator
 from types import TracebackType
-from typing import IO, Any, AnyStr, Literal, Protocol, overload
+from typing import IO, Any, AnyStr, Literal, Protocol, overload, type_check_only
 from typing_extensions import Self, TypeAlias
 
 if sys.version_info >= (3, 9):
@@ -28,9 +28,8 @@ if sys.version_info >= (3, 11):
 else:
     _TextMode: TypeAlias = Literal["r", "rU", "U"]
 
-class _HasReadlineAndFileno(Protocol[AnyStr_co]):
-    def readline(self) -> AnyStr_co: ...
-    def fileno(self) -> int: ...
+@type_check_only
+class _HasReadlineAndFileno(SupportsNoArgReadline[AnyStr_co], HasFileno, Protocol[AnyStr_co]): ...
 
 if sys.version_info >= (3, 10):
     # encoding and errors are added

stdlib/gzip.pyi

@@ -1,9 +1,9 @@
 import _compression
 import sys
 import zlib
-from _typeshed import ReadableBuffer, SizedBuffer, StrOrBytesPath
+from _typeshed import ReadableBuffer, SizedBuffer, StrOrBytesPath, SupportsFlush, SupportsWrite
 from io import FileIO
-from typing import Final, Literal, Protocol, TextIO, overload
+from typing import Final, Literal, Protocol, TextIO, overload, type_check_only
 from typing_extensions import TypeAlias
 
 __all__ = ["BadGzipFile", "GzipFile", "open", "compress", "decompress"]
@@ -21,6 +21,7 @@ FEXTRA: Final[int]  # actually Literal[4] # undocumented
 FNAME: Final[int]  # actually Literal[8] # undocumented
 FCOMMENT: Final[int]  # actually Literal[16] # undocumented
 
+@type_check_only
 class _ReadableFileobj(Protocol):
     def read(self, n: int, /) -> bytes: ...
     def seek(self, n: int, /) -> object: ...
@@ -29,13 +30,13 @@ class _ReadableFileobj(Protocol):
     # mode: str
     # def fileno() -> int: ...
 
-class _WritableFileobj(Protocol):
-    def write(self, b: bytes, /) -> object: ...
-    def flush(self) -> object: ...
+@type_check_only
+class _WritableFileobj(SupportsWrite[bytes], SupportsFlush, Protocol):
     # The following attributes and methods are optional:
     # name: str
     # mode: str
     # def fileno() -> int: ...
+    ...
 
 @overload
 def open(

stdlib/io.pyi

@@ -2,7 +2,7 @@ import abc
 import builtins
 import codecs
 import sys
-from _typeshed import FileDescriptorOrPath, ReadableBuffer, WriteableBuffer
+from _typeshed import FileDescriptorOrPath, HasFileno, ReadableBuffer, SupportsFlush, SupportsRead, SupportsWrite, WriteableBuffer
 from collections.abc import Callable, Iterable, Iterator
 from os import _Opener
 from types import TracebackType
@@ -150,24 +150,20 @@ class TextIOBase(IOBase):
     def read(self, size: int | None = ..., /) -> str: ...
 
 @type_check_only
-class _WrappedBuffer(Protocol):
+class _WrappedBuffer(SupportsRead[ReadableBuffer], SupportsWrite[bytes], SupportsFlush, HasFileno, 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.
     @property
     def name(self) -> Any: ...
     @property
     def closed(self) -> bool: ...
-    def read(self, size: int = ..., /) -> ReadableBuffer: ...
     # Optional: def read1(self, size: int, /) -> ReadableBuffer: ...
-    def write(self, b: bytes, /) -> object: ...
-    def flush(self) -> object: ...
     def close(self) -> object: ...
     def seekable(self) -> bool: ...
     def readable(self) -> bool: ...
     def writable(self) -> bool: ...
     def truncate(self, size: int, /) -> int: ...
-    def fileno(self) -> int: ...
     def isatty(self) -> bool: ...
     # Optional: Only needs to be present if seekable() returns True.
     # def seek(self, offset: Literal[0], whence: Literal[2]) -> int: ...

stdlib/pickle.pyi

@@ -1,6 +1,6 @@
-from _typeshed import ReadableBuffer, SupportsWrite
+from _typeshed import ReadableBuffer, SupportsNoArgReadline, SupportsWrite
 from collections.abc import Callable, Iterable, Iterator, Mapping
-from typing import Any, ClassVar, Protocol, SupportsBytes, SupportsIndex, final
+from typing import Any, ClassVar, Protocol, SupportsBytes, SupportsIndex, final, type_check_only
 from typing_extensions import TypeAlias
 
 __all__ = [
@@ -93,9 +93,9 @@ DEFAULT_PROTOCOL: int
 
 bytes_types: tuple[type[Any], ...]  # undocumented
 
-class _ReadableFileobj(Protocol):
+@type_check_only
+class _ReadableFileobj(SupportsNoArgReadline[bytes], Protocol):
     def read(self, n: int, /) -> bytes: ...
-    def readline(self) -> bytes: ...
 
 @final
 class PickleBuffer:

stdlib/shlex.pyi

@@ -1,4 +1,5 @@
 import sys
+from _typeshed import SupportsNoArgReadline
 from collections import deque
 from collections.abc import Iterable
 from io import TextIOWrapper
@@ -8,9 +9,8 @@ from typing_extensions import Self, deprecated
 __all__ = ["shlex", "split", "quote", "join"]
 
 @type_check_only
-class _ShlexInstream(Protocol):
+class _ShlexInstream(SupportsNoArgReadline[object], Protocol):
     def read(self, size: Literal[1], /) -> str: ...
-    def readline(self) -> object: ...
     def close(self) -> object: ...
 
 if sys.version_info >= (3, 12):

stdlib/tarfile.pyi

@@ -1,12 +1,12 @@
 import bz2
 import io
 import sys
-from _typeshed import StrOrBytesPath, StrPath
+from _typeshed import StrOrBytesPath, StrPath, SupportsWrite
 from builtins import list as _list  # aliases to avoid name clashes with fields named "type" or "list"
 from collections.abc import Callable, Iterable, Iterator, Mapping
 from gzip import _ReadableFileobj as _GzipReadableFileobj, _WritableFileobj as _GzipWritableFileobj
 from types import TracebackType
-from typing import IO, ClassVar, Literal, Protocol, overload
+from typing import IO, ClassVar, Literal, Protocol, overload, type_check_only
 from typing_extensions import Self, TypeAlias
 
 __all__ = [
@@ -42,9 +42,9 @@ if sys.version_info >= (3, 12):
 _FilterFunction: TypeAlias = Callable[[TarInfo, str], TarInfo | None]
 _TarfileFilter: TypeAlias = Literal["fully_trusted", "tar", "data"] | _FilterFunction
 
-class _Fileobj(Protocol):
+@type_check_only
+class _Fileobj(SupportsWrite[bytes], Protocol):
     def read(self, size: int, /) -> bytes: ...
-    def write(self, b: bytes, /) -> object: ...
     def tell(self) -> int: ...
     def seek(self, pos: int, /) -> object: ...
     def close(self) -> object: ...

stdlib/zipfile/__init__.pyi

@@ -1,11 +1,11 @@
 import io
 import sys
-from _typeshed import SizedBuffer, StrOrBytesPath, StrPath
+from _typeshed import SizedBuffer, StrOrBytesPath, StrPath, SupportsFlush, SupportsRead, SupportsWrite
 from collections.abc import Callable, Iterable, Iterator
 from io import TextIOWrapper
 from os import PathLike
 from types import TracebackType
-from typing import IO, Final, Literal, Protocol, overload
+from typing import IO, Final, Literal, Protocol, overload, type_check_only
 from typing_extensions import Self, TypeAlias
 
 __all__ = [
@@ -47,8 +47,8 @@ class _ZipStream(Protocol):
     # def seek(self, n: int, /) -> object: ...
 
 # Stream shape as required by _EndRecData() and _EndRecData64().
-class _SupportsReadSeekTell(Protocol):
-    def read(self, n: int = ..., /) -> bytes: ...
+@type_check_only
+class _SupportsReadSeekTell(SupportsRead[bytes], Protocol):
     def seek(self, cookie: int, whence: int, /) -> object: ...
     def tell(self) -> int: ...
 
@@ -91,22 +91,23 @@ class ZipExtFile(io.BufferedIOBase):
     def read1(self, n: int | None) -> bytes: ...  # type: ignore[override]
     def seek(self, offset: int, whence: int = 0) -> int: ...
 
-class _Writer(Protocol):
-    def write(self, s: str, /) -> object: ...
+@type_check_only
+class _Writer(SupportsWrite[str], Protocol): ...
 
-class _ZipReadable(Protocol):
+@type_check_only
+class _ZipReadable(SupportsRead[bytes], Protocol):
     def seek(self, offset: int, whence: int = 0, /) -> int: ...
-    def read(self, n: int = -1, /) -> bytes: ...
 
+@type_check_only
 class _ZipTellable(Protocol):
     def tell(self) -> int: ...
 
+@type_check_only
 class _ZipReadableTellable(_ZipReadable, _ZipTellable, Protocol): ...
 
-class _ZipWritable(Protocol):
-    def flush(self) -> None: ...
+@type_check_only
+class _ZipWritable(SupportsWrite[bytes], SupportsFlush, Protocol):
     def close(self) -> None: ...
-    def write(self, b: bytes, /) -> int: ...
 
 class ZipFile:
     filename: str | None

stubs/Flask-Migrate/flask_migrate/__init__.pyi

@@ -1,11 +1,11 @@
 # pyright: reportInvalidStubStatement=none
 
 import sys
-from _typeshed import StrPath, SupportsKeysAndGetItem, SupportsWrite
+from _typeshed import StrPath, SupportsFlush, SupportsKeysAndGetItem, SupportsWrite
 from argparse import Namespace
 from collections.abc import Callable, Iterable, Sequence
 from logging import Logger
-from typing import Any, Protocol, TypeVar
+from typing import Any, Protocol, TypeVar, type_check_only
 from typing_extensions import ParamSpec, TypeAlias
 
 import flask
@@ -20,9 +20,8 @@ _AlembicConfigValue: TypeAlias = Any
 alembic_version: tuple[int, int, int]
 log: Logger
 
-# TODO: Use _typeshed.SupportsFlush when it's available in type checkers.
-class _SupportsWriteAndFlush(SupportsWrite[_T_contra], Protocol):
-    def flush(self) -> object: ...
+@type_check_only
+class _SupportsWriteAndFlush(SupportsWrite[_T_contra], SupportsFlush, Protocol): ...
 
 class Config:  # should inherit from alembic.config.Config which is not possible yet
     template_directory: str | None

stubs/Markdown/markdown/core.pyi

@@ -1,24 +1,13 @@
+from codecs import _ReadableStream, _WritableStream
 from collections.abc import Callable, Mapping, Sequence
-from typing import Any, ClassVar, Literal, Protocol
+from typing import Any, ClassVar, Literal
 from typing_extensions import Self
 from xml.etree.ElementTree import Element
 
 from . import blockparser, inlinepatterns, postprocessors, preprocessors, treeprocessors
 from .extensions import Extension
 from .util import HtmlStash, Registry
 
-# TODO: The following protocols can be replaced by their counterparts from
-# codecs, once they have been propagated to all type checkers.
-class _WritableStream(Protocol):
-    def write(self, data: bytes, /) -> object: ...
-    def seek(self, offset: int, whence: int, /) -> object: ...
-    def close(self) -> object: ...
-
-class _ReadableStream(Protocol):
-    def read(self, size: int = ..., /) -> bytes: ...
-    def seek(self, offset: int, whence: int, /) -> object: ...
-    def close(self) -> object: ...
-
 class Markdown:
     preprocessors: Registry[preprocessors.Preprocessor]
     inlinePatterns: Registry[inlinepatterns.Pattern]

stubs/PyYAML/yaml/emitter.pyi

@@ -1,17 +1,19 @@
+from _typeshed import SupportsWrite
 from collections.abc import Callable
-from typing import Any, NoReturn, Protocol, TypeVar
+from typing import Any, NoReturn, Protocol, TypeVar, type_check_only
 
 from yaml.error import YAMLError
 
 from .events import Event
 
 _T_contra = TypeVar("_T_contra", str, bytes, contravariant=True)
 
-class _WriteStream(Protocol[_T_contra]):
-    def write(self, data: _T_contra, /) -> object: ...
+@type_check_only
+class _WriteStream(SupportsWrite[_T_contra], Protocol[_T_contra]):
     # Optional fields:
     # encoding: str
     # def flush(self) -> object: ...
+    ...
 
 class EmitterError(YAMLError): ...
 

stubs/click-spinner/click_spinner/__init__.pyi

@@ -1,14 +1,15 @@
 import threading
+from _typeshed import SupportsFlush
 from collections.abc import Iterator
 from types import TracebackType
-from typing import Literal, Protocol
+from typing import Literal, Protocol, type_check_only
 from typing_extensions import Self
 
 __version__: str
 
-class _Stream(Protocol):
+@type_check_only
+class _Stream(SupportsFlush, Protocol):
     def isatty(self) -> bool: ...
-    def flush(self) -> None: ...
     def write(self, s: str, /) -> int: ...
 
 class Spinner: