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