Add type information for asyncio.subprocess

stdlib/3.4/asyncio/__init__.pyi

@@ -19,6 +19,10 @@ from asyncio.streams import (
     IncompleteReadError as IncompleteReadError,
     LimitOverrunError as LimitOverrunError,
 )
+from asyncio.subprocess import (
+    create_subprocess_exec as create_subprocess_exec,
+    create_subprocess_shell as create_subprocess_shell,
+)
 from asyncio.transports import (
     BaseTransport as BaseTransport,
     ReadTransport as ReadTransport,
@@ -58,6 +62,7 @@ from asyncio.queues import (
 __all__ = (coroutines.__all__ +
             protocols.__all__ +
             streams.__all__ +
+            subprocess.__all__ +
             transports.__all__ +
             futures.__all__ +
             tasks.__all__ +

stdlib/3.4/asyncio/subprocess.pyi

@@ -0,0 +1,60 @@
+from typing import Any, AnyStr, Tuple
+
+__all__ = ['create_subprocess_exec', 'create_subprocess_shell']
+
+from asyncio import events
+from asyncio import protocols
+from asyncio import streams
+from asyncio import transports
+from asyncio.coroutines import coroutine
+
+
+PIPE = ... # type: int
+STDOUT = ... # type: int
+DEVNULL = ... # type: int
+
+class SubprocessStreamProtocol(streams.FlowControlMixin,
+                               protocols.SubprocessProtocol):
+    def __init__(self, limit: int, loop: events.AbstractEventLoop) -> None: ...
+    def connection_made(self, transport: transports.BaseTransport) -> None: ...
+    def pipe_data_received(self, fd: int, data: AnyStr) -> None: ...
+    def pipe_connection_lost(self, fd: int, exc: Exception): ...
+    def process_exited(self) -> None: ...
+
+
+class Process:
+    def __init__(self,
+            transport: transports.BaseTransport,
+            protocol: protocols.BaseProtocol,
+            loop: events.AbstractEventLoop) -> None: ...
+    @property
+    def returncode(self) -> int: ...
+    @coroutine
+    def wait(self) -> int: ...
+    def send_signal(self, signal: int) -> None: ...
+    def terminatate(self) -> None: ...
+    def kill(self) -> None: ...
+    @coroutine
+    def communicate(self, input: AnyStr = ...) -> Tuple[AnyStr, AnyStr]: ...
+
+
+@coroutine
+def create_subprocess_shell(
+        *Args: AnyStr,
+        stdin: int = ...,
+        stdout: int = ...,
+        stderr: int = ...,
+        loop: events.AbstractEventLoop = ...,
+        limit: int = ...,
+        **kwds: Any): ...
+
+@coroutine
+def create_subprocess_exec(
+        program: AnyStr,
+        *args: Any,
+        stdin: int = ...,
+        stdout: int = ...,
+        stderr: int = ...,
+        loop: events.AbstractEventLoop = ...,
+        limit: int = ...,
+        **kwds: Any) -> Process: ...