1 from asyncio import events
\r
2 from asyncio import protocols
\r
3 from asyncio import streams
\r
4 from asyncio import transports
\r
5 from asyncio.coroutines import coroutine
\r
6 from typing import Any, Generator, List, Optional, Text, Tuple, Union, IO
\r
10 PIPE = ... # type: int
\r
11 STDOUT = ... # type: int
\r
12 DEVNULL = ... # type: int
\r
14 class SubprocessStreamProtocol(streams.FlowControlMixin,
\r
15 protocols.SubprocessProtocol):
\r
16 stdin = ... # type: Optional[streams.StreamWriter]
\r
17 stdout = ... # type: Optional[streams.StreamReader]
\r
18 stderr = ... # type: Optional[streams.StreamReader]
\r
19 def __init__(self, limit: int, loop: events.AbstractEventLoop) -> None: ...
\r
20 def connection_made(self, transport: transports.BaseTransport) -> None: ...
\r
21 def pipe_data_received(self, fd: int, data: Union[bytes, Text]) -> None: ...
\r
22 def pipe_connection_lost(self, fd: int, exc: Exception) -> None: ...
\r
23 def process_exited(self) -> None: ...
\r
27 stdin = ... # type: Optional[streams.StreamWriter]
\r
28 stdout = ... # type: Optional[streams.StreamReader]
\r
29 stderr = ... # type: Optional[streams.StreamReader]
\r
30 pid = ... # type: int
\r
32 transport: transports.BaseTransport,
\r
33 protocol: protocols.BaseProtocol,
\r
34 loop: events.AbstractEventLoop) -> None: ...
\r
36 def returncode(self) -> int: ...
\r
38 def wait(self) -> Generator[Any, None, int]: ...
\r
39 def send_signal(self, signal: int) -> None: ...
\r
40 def terminate(self) -> None: ...
\r
41 def kill(self) -> None: ...
\r
43 def communicate(self, input: Optional[bytes] = ...) -> Generator[Any, None, Tuple[bytes, bytes]]: ...
\r
47 def create_subprocess_shell(
\r
48 *Args: Union[str, bytes], # Union used instead of AnyStr due to mypy issue #1236
\r
49 stdin: Union[int, IO[Any], None] = ...,
\r
50 stdout: Union[int, IO[Any], None] = ...,
\r
51 stderr: Union[int, IO[Any], None] = ...,
\r
52 loop: events.AbstractEventLoop = ...,
\r
55 ) -> Generator[Any, None, Process]: ...
\r
58 def create_subprocess_exec(
\r
59 program: Union[str, bytes], # Union used instead of AnyStr due to mypy issue #1236
\r
61 stdin: Union[int, IO[Any], None] = ...,
\r
62 stdout: Union[int, IO[Any], None] = ...,
\r
63 stderr: Union[int, IO[Any], None] = ...,
\r
64 loop: events.AbstractEventLoop = ...,
\r
67 ) -> Generator[Any, None, Process]: ...
\r