-
-
Notifications
You must be signed in to change notification settings - Fork 1.7k
/
streams.pyi
104 lines (94 loc) · 3.85 KB
/
streams.pyi
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
import sys
from typing import Any, AsyncIterator, Awaitable, Callable, Iterable, Optional, Tuple, Union
from . import events, protocols, transports
_ClientConnectedCallback = Callable[[StreamReader, StreamWriter], Optional[Awaitable[None]]]
if sys.version_info < (3, 8):
class IncompleteReadError(EOFError):
expected: Optional[int]
partial: bytes
def __init__(self, partial: bytes, expected: Optional[int]) -> None: ...
class LimitOverrunError(Exception):
consumed: int
def __init__(self, message: str, consumed: int) -> None: ...
async def open_connection(
host: Optional[str] = ...,
port: Optional[Union[int, str]] = ...,
*,
loop: Optional[events.AbstractEventLoop] = ...,
limit: int = ...,
ssl_handshake_timeout: Optional[float] = ...,
**kwds: Any,
) -> Tuple[StreamReader, StreamWriter]: ...
async def start_server(
client_connected_cb: _ClientConnectedCallback,
host: Optional[str] = ...,
port: Optional[Union[int, str]] = ...,
*,
loop: Optional[events.AbstractEventLoop] = ...,
limit: int = ...,
ssl_handshake_timeout: Optional[float] = ...,
**kwds: Any,
) -> events.AbstractServer: ...
if sys.platform != "win32":
if sys.version_info >= (3, 7):
from os import PathLike
_PathType = Union[str, PathLike[str]]
else:
_PathType = str
async def open_unix_connection(
path: Optional[_PathType] = ..., *, loop: Optional[events.AbstractEventLoop] = ..., limit: int = ..., **kwds: Any
) -> Tuple[StreamReader, StreamWriter]: ...
async def start_unix_server(
client_connected_cb: _ClientConnectedCallback,
path: Optional[_PathType] = ...,
*,
loop: Optional[events.AbstractEventLoop] = ...,
limit: int = ...,
**kwds: Any,
) -> events.AbstractServer: ...
class FlowControlMixin(protocols.Protocol): ...
class StreamReaderProtocol(FlowControlMixin, protocols.Protocol):
def __init__(
self,
stream_reader: StreamReader,
client_connected_cb: Optional[_ClientConnectedCallback] = ...,
loop: Optional[events.AbstractEventLoop] = ...,
) -> None: ...
def connection_made(self, transport: transports.BaseTransport) -> None: ...
def connection_lost(self, exc: Optional[Exception]) -> None: ...
def data_received(self, data: bytes) -> None: ...
def eof_received(self) -> bool: ...
class StreamWriter:
def __init__(
self,
transport: transports.BaseTransport,
protocol: protocols.BaseProtocol,
reader: Optional[StreamReader],
loop: events.AbstractEventLoop,
) -> None: ...
@property
def transport(self) -> transports.BaseTransport: ...
def write(self, data: bytes) -> None: ...
def writelines(self, data: Iterable[bytes]) -> None: ...
def write_eof(self) -> None: ...
def can_write_eof(self) -> bool: ...
def close(self) -> None: ...
if sys.version_info >= (3, 7):
def is_closing(self) -> bool: ...
async def wait_closed(self) -> None: ...
def get_extra_info(self, name: str, default: Any = ...) -> Any: ...
async def drain(self) -> None: ...
class StreamReader:
def __init__(self, limit: int = ..., loop: Optional[events.AbstractEventLoop] = ...) -> None: ...
def exception(self) -> Exception: ...
def set_exception(self, exc: Exception) -> None: ...
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: ...
async def readline(self) -> bytes: ...
async def readuntil(self, separator: bytes = ...) -> bytes: ...
async def read(self, n: int = ...) -> bytes: ...
async def readexactly(self, n: int) -> bytes: ...
def __aiter__(self) -> AsyncIterator[bytes]: ...
async def __anext__(self) -> bytes: ...