asyncio: improve bytes handling

stdlib/asyncio/base_events.pyi

@@ -1,6 +1,6 @@
 import ssl
 import sys
-from _typeshed import FileDescriptorLike, WriteableBuffer
+from _typeshed import FileDescriptorLike, ReadableBuffer, WriteableBuffer
 from asyncio.events import AbstractEventLoop, AbstractServer, Handle, TimerHandle, _TaskFactory
 from asyncio.futures import Future
 from asyncio.protocols import BaseProtocol
@@ -102,7 +102,7 @@ class BaseEventLoop(AbstractEventLoop):
     async def getaddrinfo(
         self,
         host: bytes | str | None,
-        port: str | int | None,
+        port: bytes | str | int | None,
         *,
         family: int = ...,
         type: int = ...,
@@ -411,13 +411,13 @@ class BaseEventLoop(AbstractEventLoop):
     # BaseEventLoop, only on subclasses. We list them here for now for convenience.
     async def sock_recv(self, sock: socket, nbytes: int) -> bytes: ...
     async def sock_recv_into(self, sock: socket, buf: WriteableBuffer) -> int: ...
-    async def sock_sendall(self, sock: socket, data: bytes) -> None: ...
+    async def sock_sendall(self, sock: socket, data: ReadableBuffer) -> None: ...
     async def sock_connect(self, sock: socket, address: _Address) -> None: ...
     async def sock_accept(self, sock: socket) -> tuple[socket, _RetAddress]: ...
     if sys.version_info >= (3, 11):
         async def sock_recvfrom(self, sock: socket, bufsize: int) -> bytes: ...
         async def sock_recvfrom_into(self, sock: socket, buf: WriteableBuffer, nbytes: int = ...) -> int: ...
-        async def sock_sendto(self, sock: socket, data: bytes, address: _Address) -> None: ...
+        async def sock_sendto(self, sock: socket, data: ReadableBuffer, address: _Address) -> None: ...
     # Signal handling.
     def add_signal_handler(self, sig: int, callback: Callable[..., Any], *args: Any) -> None: ...
     def remove_signal_handler(self, sig: int) -> bool: ...

stdlib/asyncio/events.pyi

@@ -1,6 +1,6 @@
 import ssl
 import sys
-from _typeshed import FileDescriptorLike, Self, StrPath, WriteableBuffer
+from _typeshed import FileDescriptorLike, ReadableBuffer, Self, StrPath, WriteableBuffer
 from abc import ABCMeta, abstractmethod
 from collections.abc import Awaitable, Callable, Coroutine, Generator, Sequence
 from contextvars import Context
@@ -194,7 +194,7 @@ class AbstractEventLoop:
     async def getaddrinfo(
         self,
         host: bytes | str | None,
-        port: str | int | None,
+        port: bytes | str | int | None,
         *,
         family: int = ...,
         type: int = ...,
@@ -562,7 +562,7 @@ class AbstractEventLoop:
     @abstractmethod
     async def sock_recv_into(self, sock: socket, buf: WriteableBuffer) -> int: ...
     @abstractmethod
-    async def sock_sendall(self, sock: socket, data: bytes) -> None: ...
+    async def sock_sendall(self, sock: socket, data: ReadableBuffer) -> None: ...
     @abstractmethod
     async def sock_connect(self, sock: socket, address: _Address) -> None: ...
     @abstractmethod
@@ -573,7 +573,7 @@ class AbstractEventLoop:
         @abstractmethod
         async def sock_recvfrom_into(self, sock: socket, buf: WriteableBuffer, nbytes: int = ...) -> int: ...
         @abstractmethod
-        async def sock_sendto(self, sock: socket, data: bytes, address: _Address) -> None: ...
+        async def sock_sendto(self, sock: socket, data: ReadableBuffer, address: _Address) -> None: ...
     # Signal handling.
     @abstractmethod
     def add_signal_handler(self, sig: int, callback: Callable[..., object], *args: Any) -> None: ...

stdlib/asyncio/sslproto.pyi

@@ -74,7 +74,7 @@ class _SSLProtocolTransport(transports._FlowControlMixin, transports.Transport):
     def get_extra_info(self, name: str, default: Any | None = ...) -> dict[str, Any]: ...
     @property
     def _protocol_paused(self) -> bool: ...
-    def write(self, data: bytes) -> None: ...
+    def write(self, data: bytes | bytearray | memoryview) -> None: ...
     def can_write_eof(self) -> Literal[False]: ...
     if sys.version_info >= (3, 11):
         def get_write_buffer_limits(self) -> tuple[int, int]: ...

stdlib/asyncio/streams.pyi

@@ -3,7 +3,7 @@ import sys
 from _typeshed import Self, StrPath
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence
 from typing import Any
-from typing_extensions import TypeAlias
+from typing_extensions import SupportsIndex, TypeAlias
 
 from . import events, protocols, transports
 from .base_events import Server
@@ -139,8 +139,8 @@ class StreamWriter:
     ) -> None: ...
     @property
     def transport(self) -> transports.WriteTransport: ...
-    def write(self, data: bytes) -> None: ...
-    def writelines(self, data: Iterable[bytes]) -> None: ...
+    def write(self, data: bytes | bytearray | memoryview) -> None: ...
+    def writelines(self, data: Iterable[bytes | bytearray | memoryview]) -> None: ...
     def write_eof(self) -> None: ...
     def can_write_eof(self) -> bool: ...
     def close(self) -> None: ...
@@ -160,9 +160,10 @@ class StreamReader(AsyncIterator[bytes]):
     def set_transport(self, transport: transports.BaseTransport) -> None: ...
     def feed_eof(self) -> None: ...
     def at_eof(self) -> bool: ...
-    def feed_data(self, data: bytes) -> None: ...
+    def feed_data(self, data: Iterable[SupportsIndex]) -> None: ...
     async def readline(self) -> bytes: ...
-    async def readuntil(self, separator: bytes = ...) -> bytes: ...
+    # Can be any buffer that supports len(); consider changing to a Protocol if PEP 688 is accepted
+    async def readuntil(self, separator: bytes | bytearray | memoryview = ...) -> bytes: ...
     async def read(self, n: int = ...) -> bytes: ...
     async def readexactly(self, n: int) -> bytes: ...
     def __aiter__(self: Self) -> Self: ...

stdlib/asyncio/subprocess.pyi

@@ -38,7 +38,7 @@ class Process:
     def send_signal(self, signal: int) -> None: ...
     def terminate(self) -> None: ...
     def kill(self) -> None: ...
-    async def communicate(self, input: bytes | None = ...) -> tuple[bytes, bytes]: ...
+    async def communicate(self, input: bytes | bytearray | memoryview | None = ...) -> tuple[bytes, bytes]: ...
 
 if sys.version_info >= (3, 10):
     async def create_subprocess_shell(

stdlib/asyncio/transports.pyi

@@ -23,16 +23,16 @@ class WriteTransport(BaseTransport):
     def set_write_buffer_limits(self, high: int | None = ..., low: int | None = ...) -> None: ...
     def get_write_buffer_size(self) -> int: ...
     def get_write_buffer_limits(self) -> tuple[int, int]: ...
-    def write(self, data: bytes) -> None: ...
-    def writelines(self, list_of_data: Iterable[bytes]) -> None: ...
+    def write(self, data: bytes | bytearray | memoryview) -> None: ...
+    def writelines(self, list_of_data: Iterable[bytes | bytearray | memoryview]) -> None: ...
     def write_eof(self) -> None: ...
     def can_write_eof(self) -> bool: ...
     def abort(self) -> None: ...
 
 class Transport(ReadTransport, WriteTransport): ...
 
 class DatagramTransport(BaseTransport):
-    def sendto(self, data: bytes, addr: _Address | None = ...) -> None: ...
+    def sendto(self, data: bytes | bytearray | memoryview, addr: _Address | None = ...) -> None: ...
     def abort(self) -> None: ...
 
 class SubprocessTransport(BaseTransport):

stdlib/asyncio/trsock.pyi

@@ -1,13 +1,14 @@
 import socket
 import sys
+from _typeshed import ReadableBuffer
 from builtins import type as Type  # alias to avoid name clashes with property named "type"
 from collections.abc import Iterable
 from types import TracebackType
 from typing import Any, BinaryIO, NoReturn, overload
 from typing_extensions import TypeAlias
 
 # These are based in socket, maybe move them out into _typeshed.pyi or such
-_Address: TypeAlias = tuple[Any, ...] | str
+_Address: TypeAlias = socket._Address
 _RetAddress: TypeAlias = Any
 _WriteBuffer: TypeAlias = bytearray | memoryview
 _CMSG: TypeAlias = tuple[int, int, bytes]
@@ -30,7 +31,7 @@ class TransportSocket:
     @overload
     def getsockopt(self, level: int, optname: int, buflen: int) -> bytes: ...
     @overload
-    def setsockopt(self, level: int, optname: int, value: int | bytes) -> None: ...
+    def setsockopt(self, level: int, optname: int, value: int | ReadableBuffer) -> None: ...
     @overload
     def setsockopt(self, level: int, optname: int, value: None, optlen: int) -> None: ...
     def getpeername(self) -> _RetAddress: ...
@@ -42,9 +43,9 @@ class TransportSocket:
     if sys.version_info < (3, 11):
         def _na(self, what: str) -> None: ...
         def accept(self) -> tuple[socket.socket, _RetAddress]: ...
-        def connect(self, address: _Address | bytes) -> None: ...
-        def connect_ex(self, address: _Address | bytes) -> int: ...
-        def bind(self, address: _Address | bytes) -> None: ...
+        def connect(self, address: _Address) -> None: ...
+        def connect_ex(self, address: _Address) -> int: ...
+        def bind(self, address: _Address) -> None: ...
         if sys.platform == "win32":
             def ioctl(self, control: int, option: int | tuple[int, int, int] | bool) -> None: ...
         else:
@@ -57,22 +58,26 @@ class TransportSocket:
         def detach(self) -> int: ...
         if sys.platform == "linux":
             def sendmsg_afalg(
-                self, msg: Iterable[bytes] = ..., *, op: int, iv: Any = ..., assoclen: int = ..., flags: int = ...
+                self, msg: Iterable[ReadableBuffer] = ..., *, op: int, iv: Any = ..., assoclen: int = ..., flags: int = ...
             ) -> int: ...
         else:
             def sendmsg_afalg(
-                self, msg: Iterable[bytes] = ..., *, op: int, iv: Any = ..., assoclen: int = ..., flags: int = ...
+                self, msg: Iterable[ReadableBuffer] = ..., *, op: int, iv: Any = ..., assoclen: int = ..., flags: int = ...
             ) -> NoReturn: ...
 
         def sendmsg(
-            self, __buffers: Iterable[bytes], __ancdata: Iterable[_CMSG] = ..., __flags: int = ..., __address: _Address = ...
+            self,
+            __buffers: Iterable[ReadableBuffer],
+            __ancdata: Iterable[_CMSG] = ...,
+            __flags: int = ...,
+            __address: _Address = ...,
         ) -> int: ...
         @overload
-        def sendto(self, data: bytes, address: _Address) -> int: ...
+        def sendto(self, data: ReadableBuffer, address: _Address) -> int: ...
         @overload
-        def sendto(self, data: bytes, flags: int, address: _Address) -> int: ...
-        def send(self, data: bytes, flags: int = ...) -> int: ...
-        def sendall(self, data: bytes, flags: int = ...) -> None: ...
+        def sendto(self, data: ReadableBuffer, flags: int, address: _Address) -> int: ...
+        def send(self, data: ReadableBuffer, flags: int = ...) -> int: ...
+        def sendall(self, data: ReadableBuffer, flags: int = ...) -> None: ...
         def set_inheritable(self, inheritable: bool) -> None: ...
         if sys.platform == "win32":
             def share(self, process_id: int) -> bytes: ...

stdlib/asyncio/windows_events.pyi

@@ -1,6 +1,6 @@
 import socket
 import sys
-from _typeshed import WriteableBuffer
+from _typeshed import Incomplete, WriteableBuffer
 from collections.abc import Callable
 from typing import IO, Any, ClassVar, NoReturn
 from typing_extensions import Literal
@@ -50,10 +50,14 @@ if sys.platform == "win32":
         def recv_into(self, conn: socket.socket, buf: WriteableBuffer, flags: int = ...) -> futures.Future[Any]: ...
         def send(self, conn: socket.socket, buf: WriteableBuffer, flags: int = ...) -> futures.Future[Any]: ...
         def accept(self, listener: socket.socket) -> futures.Future[Any]: ...
-        def connect(self, conn: socket.socket, address: bytes) -> futures.Future[Any]: ...
+        def connect(
+            self,
+            conn: socket.socket,
+            address: tuple[Incomplete, Incomplete] | tuple[Incomplete, Incomplete, Incomplete, Incomplete],
+        ) -> futures.Future[Any]: ...
         def sendfile(self, sock: socket.socket, file: IO[bytes], offset: int, count: int) -> futures.Future[Any]: ...
         def accept_pipe(self, pipe: socket.socket) -> futures.Future[Any]: ...
-        async def connect_pipe(self, address: bytes) -> windows_utils.PipeHandle: ...
+        async def connect_pipe(self, address: str) -> windows_utils.PipeHandle: ...
         def wait_for_handle(self, handle: windows_utils.PipeHandle, timeout: int | None = ...) -> bool: ...
         def close(self) -> None: ...
     SelectorEventLoop = _WindowsSelectorEventLoop