VimPlug/typeshed
1
0
Fork 0
mirror of https://github.com/davidhalter/typeshed.git synced 2025-12-16 00:37:10 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Add missing stubs for multiprocessing.connection module. (#2596)

Browse Source
This commit is contained in:
Dev Aggarwal
2018-11-11 21:24:11 +05:30
committed by Sebastian Rittau
parent f8e5222c46
commit c7b0e60032
1 changed files with 29 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
stdlib/3/multiprocessing/connection.pyi
View File

@@ -1,15 +1,34 @@
from typing import Any, Iterable, List, Optional, Tuple, Type, Union
import socket
import sys
import types
# https://docs.python.org/3/library/multiprocessing.html#address-formats
_Address = Union[str, Tuple[str, int]]
def deliver_challenge(connection: Connection, authkey: bytes) -> None: ...
def answer_challenge(connection: Connection, authkey: bytes) -> None: ...
def wait(object_list: Iterable[Union[Connection, socket.socket, int]], timeout: Optional[float] = ...) -> List[Union[Connection, socket.socket, int]]: ...
def Client(address: _Address, family: Optional[str] = ..., authkey: Optional[bytes] = ...) -> Connection: ...
def Pipe(duplex: bool = ...) -> Tuple[Connection, Connection]: ...
class _ConnectionBase:
@property
def closed(self) -> bool: ... # undocumented
@property
def readable(self) -> bool: ... # undocumented
@property
def writable(self) -> bool: ... # undocumented
def fileno(self) -> int: ...
def close(self) -> None: ...
def send_bytes(self,
buf: bytes,
offset: int = ...,
size: Optional[int] = ...) -> None: ...
def send(self, obj: Any) -> None: ...
def recv_bytes(self, maxlength: Optional[int] = ...) -> bytes: ...
def recv_bytes_into(self, buf: Any, offset: int = ...) -> int: ...
def recv(self) -> Any: ...
def poll(self, timeout: Optional[float] = ...) -> bool: ...
class Connection(_ConnectionBase): ...
if sys.platform == "win32":
class PipeConnection(_ConnectionBase): ...
class Listener:
def __init__(self, address: Optional[_Address] = ..., family: Optional[str] = ..., backlog: int = ..., authkey: Optional[bytes] = ...) -> None: ...
@@ -22,15 +41,8 @@ class Listener:
def __enter__(self) -> Listener: ...
def __exit__(self, exc_type: Optional[Type[BaseException]], exc_value: Optional[BaseException], exc_tb: Optional[types.TracebackType]) -> None: ...
class Connection:
def close(self) -> None: ...
def fileno(self) -> int: ...
def poll(self, timeout: Optional[float] = ...) -> bool: ...
def recv(self) -> Any: ...
def recv_bytes(self, maxlength: Optional[int] = ...) -> bytes: ...
def recv_bytes_into(self, buf: Any, offset: int = ...) -> int: ...
def send(self, obj: Any) -> None: ...
def send_bytes(self,
buf: bytes,
offset: int = ...,
size: Optional[int] = ...) -> None: ...
def deliver_challenge(connection: Connection, authkey: bytes) -> None: ...
def answer_challenge(connection: Connection, authkey: bytes) -> None: ...
def wait(object_list: Iterable[Union[Connection, socket.socket, int]], timeout: Optional[float] = ...) -> List[Union[Connection, socket.socket, int]]: ...
def Client(address: _Address, family: Optional[str] = ..., authkey: Optional[bytes] = ...) -> Connection: ...
def Pipe(duplex: bool = ...) -> Tuple[Connection, Connection]: ...