--- /dev/null
+from asyncio import events\r
+from asyncio import protocols\r
+from asyncio import streams\r
+from asyncio import transports\r
+from asyncio.coroutines import coroutine\r
+from typing import Any, Generator, List, Optional, Text, Tuple, Union, IO\r
+\r
+__all__: List[str]\r
+\r
+PIPE = ... # type: int\r
+STDOUT = ... # type: int\r
+DEVNULL = ... # type: int\r
+\r
+class SubprocessStreamProtocol(streams.FlowControlMixin,\r
+ protocols.SubprocessProtocol):\r
+ stdin = ... # type: Optional[streams.StreamWriter]\r
+ stdout = ... # type: Optional[streams.StreamReader]\r
+ stderr = ... # type: Optional[streams.StreamReader]\r
+ def __init__(self, limit: int, loop: events.AbstractEventLoop) -> None: ...\r
+ def connection_made(self, transport: transports.BaseTransport) -> None: ...\r
+ def pipe_data_received(self, fd: int, data: Union[bytes, Text]) -> None: ...\r
+ def pipe_connection_lost(self, fd: int, exc: Exception) -> None: ...\r
+ def process_exited(self) -> None: ...\r
+\r
+\r
+class Process:\r
+ stdin = ... # type: Optional[streams.StreamWriter]\r
+ stdout = ... # type: Optional[streams.StreamReader]\r
+ stderr = ... # type: Optional[streams.StreamReader]\r
+ pid = ... # type: int\r
+ def __init__(self,\r
+ transport: transports.BaseTransport,\r
+ protocol: protocols.BaseProtocol,\r
+ loop: events.AbstractEventLoop) -> None: ...\r
+ @property\r
+ def returncode(self) -> int: ...\r
+ @coroutine\r
+ def wait(self) -> Generator[Any, None, int]: ...\r
+ def send_signal(self, signal: int) -> None: ...\r
+ def terminate(self) -> None: ...\r
+ def kill(self) -> None: ...\r
+ @coroutine\r
+ def communicate(self, input: Optional[bytes] = ...) -> Generator[Any, None, Tuple[bytes, bytes]]: ...\r
+\r
+\r
+@coroutine\r
+def create_subprocess_shell(\r
+ *Args: Union[str, bytes], # Union used instead of AnyStr due to mypy issue #1236\r
+ stdin: Union[int, IO[Any], None] = ...,\r
+ stdout: Union[int, IO[Any], None] = ...,\r
+ stderr: Union[int, IO[Any], None] = ...,\r
+ loop: events.AbstractEventLoop = ...,\r
+ limit: int = ...,\r
+ **kwds: Any\r
+) -> Generator[Any, None, Process]: ...\r
+\r
+@coroutine\r
+def create_subprocess_exec(\r
+ program: Union[str, bytes], # Union used instead of AnyStr due to mypy issue #1236\r
+ *args: Any,\r
+ stdin: Union[int, IO[Any], None] = ...,\r
+ stdout: Union[int, IO[Any], None] = ...,\r
+ stderr: Union[int, IO[Any], None] = ...,\r
+ loop: events.AbstractEventLoop = ...,\r
+ limit: int = ...,\r
+ **kwds: Any\r
+) -> Generator[Any, None, Process]: ...\r