VimPlug/typeshed
1
0
Fork 0
mirror of https://github.com/davidhalter/typeshed.git synced 2026-01-28 13:52:12 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

asyncio: improve bytes handling (#9013)

Browse Source
This commit is contained in:
Jelle Zijlstra
2022-10-28 15:36:44 -07:00
committed by GitHub
parent b8659e69f5
commit 583b600db0
8 changed files with 43 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
stdlib/asyncio/trsock.pyi
View File

@@ -1,5 +1,6 @@
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
@@ -7,7 +8,7 @@ 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: ...