1import subprocess 2import sys 3from _typeshed import StrOrBytesPath 4from asyncio import events, protocols, streams, transports 5from typing import IO, Any, Callable, Optional, Tuple, Union 6from typing_extensions import Literal 7 8if sys.version_info >= (3, 8): 9 _ExecArg = StrOrBytesPath 10else: 11 _ExecArg = Union[str, bytes] 12 13PIPE: int 14STDOUT: int 15DEVNULL: int 16 17class SubprocessStreamProtocol(streams.FlowControlMixin, protocols.SubprocessProtocol): 18 stdin: Optional[streams.StreamWriter] 19 stdout: Optional[streams.StreamReader] 20 stderr: Optional[streams.StreamReader] 21 def __init__(self, limit: int, loop: events.AbstractEventLoop) -> None: ... 22 def connection_made(self, transport: transports.BaseTransport) -> None: ... 23 def pipe_data_received(self, fd: int, data: Union[bytes, str]) -> None: ... 24 def pipe_connection_lost(self, fd: int, exc: Optional[Exception]) -> None: ... 25 def process_exited(self) -> None: ... 26 27class Process: 28 stdin: Optional[streams.StreamWriter] 29 stdout: Optional[streams.StreamReader] 30 stderr: Optional[streams.StreamReader] 31 pid: int 32 def __init__( 33 self, transport: transports.BaseTransport, protocol: protocols.BaseProtocol, loop: events.AbstractEventLoop 34 ) -> None: ... 35 @property 36 def returncode(self) -> Optional[int]: ... 37 async def wait(self) -> int: ... 38 def send_signal(self, signal: int) -> None: ... 39 def terminate(self) -> None: ... 40 def kill(self) -> None: ... 41 async def communicate(self, input: Optional[bytes] = ...) -> Tuple[bytes, bytes]: ... 42 43if sys.version_info >= (3, 10): 44 async def create_subprocess_shell( 45 cmd: Union[str, bytes], 46 stdin: Union[int, IO[Any], None] = ..., 47 stdout: Union[int, IO[Any], None] = ..., 48 stderr: Union[int, IO[Any], None] = ..., 49 limit: int = ..., 50 *, 51 # These parameters are forced to these values by BaseEventLoop.subprocess_shell 52 universal_newlines: Literal[False] = ..., 53 shell: Literal[True] = ..., 54 bufsize: Literal[0] = ..., 55 encoding: None = ..., 56 errors: None = ..., 57 text: Literal[False, None] = ..., 58 # These parameters are taken by subprocess.Popen, which this ultimately delegates to 59 executable: Optional[StrOrBytesPath] = ..., 60 preexec_fn: Optional[Callable[[], Any]] = ..., 61 close_fds: bool = ..., 62 cwd: Optional[StrOrBytesPath] = ..., 63 env: Optional[subprocess._ENV] = ..., 64 startupinfo: Optional[Any] = ..., 65 creationflags: int = ..., 66 restore_signals: bool = ..., 67 start_new_session: bool = ..., 68 pass_fds: Any = ..., 69 ) -> Process: ... 70 async def create_subprocess_exec( 71 program: _ExecArg, 72 *args: _ExecArg, 73 stdin: Union[int, IO[Any], None] = ..., 74 stdout: Union[int, IO[Any], None] = ..., 75 stderr: Union[int, IO[Any], None] = ..., 76 limit: int = ..., 77 # These parameters are forced to these values by BaseEventLoop.subprocess_shell 78 universal_newlines: Literal[False] = ..., 79 shell: Literal[True] = ..., 80 bufsize: Literal[0] = ..., 81 encoding: None = ..., 82 errors: None = ..., 83 # These parameters are taken by subprocess.Popen, which this ultimately delegates to 84 text: Optional[bool] = ..., 85 executable: Optional[StrOrBytesPath] = ..., 86 preexec_fn: Optional[Callable[[], Any]] = ..., 87 close_fds: bool = ..., 88 cwd: Optional[StrOrBytesPath] = ..., 89 env: Optional[subprocess._ENV] = ..., 90 startupinfo: Optional[Any] = ..., 91 creationflags: int = ..., 92 restore_signals: bool = ..., 93 start_new_session: bool = ..., 94 pass_fds: Any = ..., 95 ) -> Process: ... 96 97else: 98 async def create_subprocess_shell( 99 cmd: Union[str, bytes], 100 stdin: Union[int, IO[Any], None] = ..., 101 stdout: Union[int, IO[Any], None] = ..., 102 stderr: Union[int, IO[Any], None] = ..., 103 loop: Optional[events.AbstractEventLoop] = ..., 104 limit: int = ..., 105 *, 106 # These parameters are forced to these values by BaseEventLoop.subprocess_shell 107 universal_newlines: Literal[False] = ..., 108 shell: Literal[True] = ..., 109 bufsize: Literal[0] = ..., 110 encoding: None = ..., 111 errors: None = ..., 112 text: Literal[False, None] = ..., 113 # These parameters are taken by subprocess.Popen, which this ultimately delegates to 114 executable: Optional[StrOrBytesPath] = ..., 115 preexec_fn: Optional[Callable[[], Any]] = ..., 116 close_fds: bool = ..., 117 cwd: Optional[StrOrBytesPath] = ..., 118 env: Optional[subprocess._ENV] = ..., 119 startupinfo: Optional[Any] = ..., 120 creationflags: int = ..., 121 restore_signals: bool = ..., 122 start_new_session: bool = ..., 123 pass_fds: Any = ..., 124 ) -> Process: ... 125 async def create_subprocess_exec( 126 program: _ExecArg, 127 *args: _ExecArg, 128 stdin: Union[int, IO[Any], None] = ..., 129 stdout: Union[int, IO[Any], None] = ..., 130 stderr: Union[int, IO[Any], None] = ..., 131 loop: Optional[events.AbstractEventLoop] = ..., 132 limit: int = ..., 133 # These parameters are forced to these values by BaseEventLoop.subprocess_shell 134 universal_newlines: Literal[False] = ..., 135 shell: Literal[True] = ..., 136 bufsize: Literal[0] = ..., 137 encoding: None = ..., 138 errors: None = ..., 139 # These parameters are taken by subprocess.Popen, which this ultimately delegates to 140 text: Optional[bool] = ..., 141 executable: Optional[StrOrBytesPath] = ..., 142 preexec_fn: Optional[Callable[[], Any]] = ..., 143 close_fds: bool = ..., 144 cwd: Optional[StrOrBytesPath] = ..., 145 env: Optional[subprocess._ENV] = ..., 146 startupinfo: Optional[Any] = ..., 147 creationflags: int = ..., 148 restore_signals: bool = ..., 149 start_new_session: bool = ..., 150 pass_fds: Any = ..., 151 ) -> Process: ... 152