Use PEP 688 (#10225)

Co-authored-by: Alex Waygood <Alex.Waygood@Gmail.com>

stdlib/_ctypes.pyi

@@ -63,6 +63,8 @@ class _CData(metaclass=_CDataMeta):
     def from_param(cls, obj: Any) -> Self | _CArgObject: ...
     @classmethod
     def in_dll(cls, library: CDLL, name: str) -> Self: ...
+    def __buffer__(self, __flags: int) -> memoryview: ...
+    def __release_buffer__(self, __buffer: memoryview) -> None: ...
 
 class _SimpleCData(Generic[_T], _CData):
     value: _T

stdlib/_typeshed/__init__.pyi

@@ -2,17 +2,13 @@
 #
 # See the README.md file in this directory for more information.
 
-import array
-import ctypes
-import mmap
-import pickle
 import sys
-from collections.abc import Awaitable, Callable, Iterable, Set as AbstractSet
+from collections.abc import Awaitable, Callable, Iterable, Sequence, Set as AbstractSet, Sized
 from dataclasses import Field
 from os import PathLike
 from types import FrameType, TracebackType
-from typing import Any, AnyStr, ClassVar, Generic, Protocol, TypeVar
-from typing_extensions import Final, Literal, LiteralString, TypeAlias, final
+from typing import Any, AnyStr, ClassVar, Generic, Protocol, TypeVar, overload
+from typing_extensions import Buffer, Final, Literal, LiteralString, TypeAlias, final
 
 _KT = TypeVar("_KT")
 _KT_co = TypeVar("_KT_co", covariant=True)
@@ -227,42 +223,33 @@ class SupportsNoArgReadline(Protocol[_T_co]):
 class SupportsWrite(Protocol[_T_contra]):
     def write(self, __s: _T_contra) -> object: ...
 
-ReadOnlyBuffer: TypeAlias = bytes  # stable
+# Unfortunately PEP 688 does not allow us to distinguish read-only
+# from writable buffers. We use these aliases for readability for now.
+# Perhaps a future extension of the buffer protocol will allow us to
+# distinguish these cases in the type system.
+ReadOnlyBuffer: TypeAlias = Buffer  # stable
 # Anything that implements the read-write buffer interface.
-# The buffer interface is defined purely on the C level, so we cannot define a normal Protocol
-# for it (until PEP 688 is implemented). Instead we have to list the most common stdlib buffer classes in a Union.
-if sys.version_info >= (3, 8):
-    WriteableBuffer: TypeAlias = (
-        bytearray | memoryview | array.array[Any] | mmap.mmap | ctypes._CData | pickle.PickleBuffer
-    )  # stable
-else:
-    WriteableBuffer: TypeAlias = bytearray | memoryview | array.array[Any] | mmap.mmap | ctypes._CData  # stable
-# Same as _WriteableBuffer, but also includes read-only buffer types (like bytes).
-ReadableBuffer: TypeAlias = ReadOnlyBuffer | WriteableBuffer  # stable
-_BufferWithLen: TypeAlias = ReadableBuffer  # not stable  # noqa: Y047
+WriteableBuffer: TypeAlias = Buffer
+# Same as WriteableBuffer, but also includes read-only buffer types (like bytes).
+ReadableBuffer: TypeAlias = Buffer  # stable
 
-# Anything that implements the read-write buffer interface, and can be sliced/indexed.
-SliceableBuffer: TypeAlias = bytes | bytearray | memoryview | array.array[Any] | mmap.mmap
-IndexableBuffer: TypeAlias = bytes | bytearray | memoryview | array.array[Any] | mmap.mmap
-# https://github.com/python/typeshed/pull/9115#issuecomment-1304905864
-# Post PEP 688, they should be rewritten as such:
-# from collections.abc import Sequence
-# from typing import Sized, overload
-# class SliceableBuffer(Protocol):
-#     def __buffer__(self, __flags: int) -> memoryview: ...
-#     def __getitem__(self, __slice: slice) -> Sequence[int]: ...
-# class IndexableBuffer(Protocol):
-#     def __buffer__(self, __flags: int) -> memoryview: ...
-#     def __getitem__(self, __i: int) -> int: ...
-# class SupportsGetItemBuffer(SliceableBuffer, IndexableBuffer, Protocol):
-#     def __buffer__(self, __flags: int) -> memoryview: ...
-#     def __contains__(self, __x: Any) -> bool: ...
-#     @overload
-#     def __getitem__(self, __slice: slice) -> Sequence[int]: ...
-#     @overload
-#     def __getitem__(self, __i: int) -> int: ...
-# class SizedBuffer(Sized, Protocol):  # instead of _BufferWithLen
-#     def __buffer__(self, __flags: int) -> memoryview: ...
+class SliceableBuffer(Buffer, Protocol):
+    def __getitem__(self, __slice: slice) -> Sequence[int]: ...
+
+class IndexableBuffer(Buffer, Protocol):
+    def __getitem__(self, __i: int) -> int: ...
+
+class SupportsGetItemBuffer(SliceableBuffer, IndexableBuffer, Protocol):
+    def __contains__(self, __x: Any) -> bool: ...
+    @overload
+    def __getitem__(self, __slice: slice) -> Sequence[int]: ...
+    @overload
+    def __getitem__(self, __i: int) -> int: ...
+
+class SizedBuffer(Sized, Buffer, Protocol): ...
+
+# for compatibility with third-party stubs that may use this
+_BufferWithLen: TypeAlias = SizedBuffer  # not stable  # noqa: Y047
 
 ExcInfo: TypeAlias = tuple[type[BaseException], BaseException, TracebackType]
 OptExcInfo: TypeAlias = ExcInfo | tuple[None, None, None]

stdlib/array.pyi

@@ -80,5 +80,7 @@ class array(MutableSequence[_T], Generic[_T]):
     def __rmul__(self, __value: int) -> array[_T]: ...
     def __copy__(self) -> array[_T]: ...
     def __deepcopy__(self, __unused: Any) -> array[_T]: ...
+    def __buffer__(self, __flags: int) -> memoryview: ...
+    def __release_buffer__(self, __buffer: memoryview) -> None: ...
 
 ArrayType = array

stdlib/binhex.pyi

@@ -1,4 +1,4 @@
-from _typeshed import _BufferWithLen
+from _typeshed import SizedBuffer
 from typing import IO, Any
 from typing_extensions import Literal, TypeAlias
 
@@ -28,9 +28,9 @@ class openrsrc:
 
 class BinHex:
     def __init__(self, name_finfo_dlen_rlen: _FileInfoTuple, ofp: _FileHandleUnion) -> None: ...
-    def write(self, data: _BufferWithLen) -> None: ...
+    def write(self, data: SizedBuffer) -> None: ...
     def close_data(self) -> None: ...
-    def write_rsrc(self, data: _BufferWithLen) -> None: ...
+    def write_rsrc(self, data: SizedBuffer) -> None: ...
     def close(self) -> None: ...
 
 def binhex(inp: str, out: str) -> None: ...

stdlib/builtins.pyi

@@ -695,6 +695,8 @@ class bytes(ByteString):
     if sys.version_info >= (3, 11):
         def __bytes__(self) -> bytes: ...
 
+    def __buffer__(self, __flags: int) -> memoryview: ...
+
 class bytearray(MutableSequence[int], ByteString):
     @overload
     def __init__(self) -> None: ...
@@ -810,6 +812,8 @@ class bytearray(MutableSequence[int], ByteString):
     def __gt__(self, __value: ReadableBuffer) -> bool: ...
     def __ge__(self, __value: ReadableBuffer) -> bool: ...
     def __alloc__(self) -> int: ...
+    def __buffer__(self, __flags: int) -> memoryview: ...
+    def __release_buffer__(self, __buffer: memoryview) -> None: ...
 
 @final
 class memoryview(Sequence[int]):
@@ -871,6 +875,9 @@ class memoryview(Sequence[int]):
     else:
         def hex(self) -> str: ...
 
+    def __buffer__(self, __flags: int) -> memoryview: ...
+    def __release_buffer__(self, __buffer: memoryview) -> None: ...
+
 @final
 class bool(int):
     def __new__(cls, __o: object = ...) -> Self: ...

stdlib/fcntl.pyi

@@ -1,7 +1,7 @@
 import sys
 from _typeshed import FileDescriptorLike, ReadOnlyBuffer, WriteableBuffer
 from typing import Any, overload
-from typing_extensions import Literal
+from typing_extensions import Buffer, Literal
 
 if sys.platform != "win32":
     FASYNC: int
@@ -104,13 +104,19 @@ if sys.platform != "win32":
     def fcntl(__fd: FileDescriptorLike, __cmd: int, __arg: int = 0) -> int: ...
     @overload
     def fcntl(__fd: FileDescriptorLike, __cmd: int, __arg: str | ReadOnlyBuffer) -> bytes: ...
+    # If arg is an int, return int
     @overload
     def ioctl(__fd: FileDescriptorLike, __request: int, __arg: int = 0, __mutate_flag: bool = True) -> int: ...
+    # The return type works as follows:
+    # - If arg is a read-write buffer, return int if mutate_flag is True, otherwise bytes
+    # - If arg is a read-only buffer, return bytes (and ignore the value of mutate_flag)
+    # We can't represent that precisely as we can't distinguish between read-write and read-only
+    # buffers, so we add overloads for a few unambiguous cases and use Any for the rest.
     @overload
-    def ioctl(__fd: FileDescriptorLike, __request: int, __arg: WriteableBuffer, __mutate_flag: Literal[True] = True) -> int: ...
+    def ioctl(__fd: FileDescriptorLike, __request: int, __arg: bytes, __mutate_flag: bool = True) -> bytes: ...
     @overload
     def ioctl(__fd: FileDescriptorLike, __request: int, __arg: WriteableBuffer, __mutate_flag: Literal[False]) -> bytes: ...
     @overload
-    def ioctl(__fd: FileDescriptorLike, __request: int, __arg: ReadOnlyBuffer, __mutate_flag: bool = True) -> bytes: ...
+    def ioctl(__fd: FileDescriptorLike, __request: int, __arg: Buffer, __mutate_flag: bool = True) -> Any: ...
     def flock(__fd: FileDescriptorLike, __operation: int) -> None: ...
     def lockf(__fd: FileDescriptorLike, __cmd: int, __len: int = 0, __start: int = 0, __whence: int = 0) -> Any: ...

stdlib/gzip.pyi

@@ -1,7 +1,7 @@
 import _compression
 import sys
 import zlib
-from _typeshed import ReadableBuffer, StrOrBytesPath, _BufferWithLen
+from _typeshed import ReadableBuffer, SizedBuffer, StrOrBytesPath
 from io import FileIO
 from typing import Protocol, TextIO, overload
 from typing_extensions import Literal, TypeAlias
@@ -159,9 +159,9 @@ class _GzipReader(_compression.DecompressReader):
     def __init__(self, fp: _ReadableFileobj) -> None: ...
 
 if sys.version_info >= (3, 8):
-    def compress(data: _BufferWithLen, compresslevel: int = 9, *, mtime: float | None = None) -> bytes: ...
+    def compress(data: SizedBuffer, compresslevel: int = 9, *, mtime: float | None = None) -> bytes: ...
 
 else:
-    def compress(data: _BufferWithLen, compresslevel: int = 9) -> bytes: ...
+    def compress(data: SizedBuffer, compresslevel: int = 9) -> bytes: ...
 
 def decompress(data: ReadableBuffer) -> bytes: ...

stdlib/hmac.pyi

@@ -1,5 +1,5 @@
 import sys
-from _typeshed import ReadableBuffer, _BufferWithLen
+from _typeshed import ReadableBuffer, SizedBuffer
 from collections.abc import Callable
 from types import ModuleType
 from typing import Any, AnyStr, overload
@@ -46,4 +46,4 @@ class HMAC:
 def compare_digest(__a: ReadableBuffer, __b: ReadableBuffer) -> bool: ...
 @overload
 def compare_digest(__a: AnyStr, __b: AnyStr) -> bool: ...
-def digest(key: _BufferWithLen, msg: ReadableBuffer, digest: _DigestMod) -> bytes: ...
+def digest(key: SizedBuffer, msg: ReadableBuffer, digest: _DigestMod) -> bytes: ...

stdlib/imaplib.pyi

@@ -1,7 +1,7 @@
 import subprocess
 import sys
 import time
-from _typeshed import ReadableBuffer, _BufferWithLen
+from _typeshed import ReadableBuffer, SizedBuffer
 from builtins import list as _list  # conflicts with a method named "list"
 from collections.abc import Callable
 from datetime import datetime
@@ -155,7 +155,7 @@ class _Authenticator:
     def __init__(self, mechinst: Callable[[bytes], bytes | bytearray | memoryview | str | None]) -> None: ...
     def process(self, data: str) -> str: ...
     def encode(self, inp: bytes | bytearray | memoryview) -> str: ...
-    def decode(self, inp: str | _BufferWithLen) -> bytes: ...
+    def decode(self, inp: str | SizedBuffer) -> bytes: ...
 
 def Internaldate2tuple(resp: ReadableBuffer) -> time.struct_time | None: ...
 def Int2AP(num: SupportsAbs[SupportsInt]) -> bytes: ...

stdlib/mmap.pyi

@@ -76,6 +76,8 @@ class mmap(Iterable[int], Sized):
     def __iter__(self) -> Iterator[int]: ...
     def __enter__(self) -> Self: ...
     def __exit__(self, *args: Unused) -> None: ...
+    def __buffer__(self, __flags: int) -> memoryview: ...
+    def __release_buffer__(self, __buffer: memoryview) -> None: ...
 
 if sys.version_info >= (3, 8) and sys.platform != "win32":
     MADV_NORMAL: int

stdlib/pickle.pyi

@@ -103,6 +103,8 @@ if sys.version_info >= (3, 8):
         def __init__(self, buffer: ReadableBuffer) -> None: ...
         def raw(self) -> memoryview: ...
         def release(self) -> None: ...
+        def __buffer__(self, __flags: int) -> memoryview: ...
+        def __release_buffer__(self, __buffer: memoryview) -> None: ...
     _BufferCallback: TypeAlias = Callable[[PickleBuffer], Any] | None
     def dump(
         obj: Any,

stdlib/smtplib.pyi

@@ -1,6 +1,6 @@
 import sys
 from _socket import _Address as _SourceAddress
-from _typeshed import ReadableBuffer, _BufferWithLen
+from _typeshed import ReadableBuffer, SizedBuffer
 from collections.abc import Sequence
 from email.message import Message as _Message
 from re import Pattern
@@ -133,7 +133,7 @@ class SMTP:
         self,
         from_addr: str,
         to_addrs: str | Sequence[str],
-        msg: _BufferWithLen | str,
+        msg: SizedBuffer | str,
         mail_options: Sequence[str] = (),
         rcpt_options: Sequence[str] = (),
     ) -> _SendErrs: ...

stdlib/typing_extensions.pyi

@@ -398,4 +398,8 @@ else:
             def __or__(self, right: Any) -> _SpecialForm: ...
             def __ror__(self, left: Any) -> _SpecialForm: ...
 
-    class Buffer(abc.ABC): ...
+    @runtime_checkable
+    class Buffer(Protocol):
+        # Not actually a Protocol at runtime; see
+        # https://github.com/python/typeshed/issues/10224 for why we're defining it this way
+        def __buffer__(self, __flags: int) -> memoryview: ...

stdlib/xmlrpc/client.pyi

@@ -2,7 +2,7 @@ import gzip
 import http.client
 import sys
 import time
-from _typeshed import ReadableBuffer, SupportsRead, SupportsWrite, _BufferWithLen
+from _typeshed import ReadableBuffer, SizedBuffer, SupportsRead, SupportsWrite
 from collections.abc import Callable, Iterable, Mapping
 from datetime import datetime
 from io import BytesIO
@@ -236,20 +236,20 @@ class Transport:
         def __init__(self, use_datetime: bool = False, use_builtin_types: bool = False) -> None: ...
 
     def request(
-        self, host: _HostType, handler: str, request_body: _BufferWithLen, verbose: bool = False
+        self, host: _HostType, handler: str, request_body: SizedBuffer, verbose: bool = False
     ) -> tuple[_Marshallable, ...]: ...
     def single_request(
-        self, host: _HostType, handler: str, request_body: _BufferWithLen, verbose: bool = False
+        self, host: _HostType, handler: str, request_body: SizedBuffer, verbose: bool = False
     ) -> tuple[_Marshallable, ...]: ...
     def getparser(self) -> tuple[ExpatParser, Unmarshaller]: ...
     def get_host_info(self, host: _HostType) -> tuple[str, list[tuple[str, str]], dict[str, str]]: ...
     def make_connection(self, host: _HostType) -> http.client.HTTPConnection: ...
     def close(self) -> None: ...
     def send_request(
-        self, host: _HostType, handler: str, request_body: _BufferWithLen, debug: bool
+        self, host: _HostType, handler: str, request_body: SizedBuffer, debug: bool
     ) -> http.client.HTTPConnection: ...
     def send_headers(self, connection: http.client.HTTPConnection, headers: list[tuple[str, str]]) -> None: ...
-    def send_content(self, connection: http.client.HTTPConnection, request_body: _BufferWithLen) -> None: ...
+    def send_content(self, connection: http.client.HTTPConnection, request_body: SizedBuffer) -> None: ...
     def parse_response(self, response: http.client.HTTPResponse) -> tuple[_Marshallable, ...]: ...
 
 class SafeTransport(Transport):

stdlib/zipfile.pyi

@@ -1,6 +1,6 @@
 import io
 import sys
-from _typeshed import StrOrBytesPath, StrPath, _BufferWithLen
+from _typeshed import SizedBuffer, StrOrBytesPath, StrPath
 from collections.abc import Callable, Iterable, Iterator
 from os import PathLike
 from types import TracebackType
@@ -179,7 +179,7 @@ class ZipFile:
     def writestr(
         self,
         zinfo_or_arcname: str | ZipInfo,
-        data: _BufferWithLen | str,
+        data: SizedBuffer | str,
         compress_type: int | None = None,
         compresslevel: int | None = None,
     ) -> None: ...

tests/stubtest_allowlists/py3_common.txt

@@ -774,3 +774,7 @@ pkgutil.ImpLoader.get_source
 pkgutil.ImpLoader.is_package
 pkgutil.ImpLoader.load_module
 pkgutil.ImpLoader.source
+
+# We lie about the existence of these methods
+.*.__buffer__
+.*.__release_buffer__