typeshed/stdlib/asyncio/subprocess.pyi at 5ebbe62215e4085e2f0b80d66f09adb4e2550e63 · JelleZijlstra/typeshed · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
import subprocess
import sys
from _typeshed import StrOrBytesPath
from asyncio import events, protocols, streams, transports
from collections.abc import Callable
from typing import IO, Any
from typing_extensions import Literal, TypeAlias

__all__ = ("create_subprocess_exec", "create_subprocess_shell")

if sys.version_info >= (3, 8):
    _ExecArg: TypeAlias = StrOrBytesPath
else:
    _ExecArg: TypeAlias = str | bytes

PIPE: int
STDOUT: int
DEVNULL: int

class SubprocessStreamProtocol(streams.FlowControlMixin, protocols.SubprocessProtocol):
    stdin: streams.StreamWriter | None
    stdout: streams.StreamReader | None
    stderr: streams.StreamReader | None
    def __init__(self, limit: int, loop: events.AbstractEventLoop) -> None: ...
    def pipe_data_received(self, fd: int, data: bytes | str) -> None: ...

class Process:
    stdin: streams.StreamWriter | None
    stdout: streams.StreamReader | None
    stderr: streams.StreamReader | None
    pid: int
    def __init__(
        self, transport: transports.BaseTransport, protocol: protocols.BaseProtocol, loop: events.AbstractEventLoop
    ) -> None: ...
    @property
    def returncode(self) -> int | None: ...
    async def wait(self) -> int: ...
    def send_signal(self, signal: int) -> None: ...
    def terminate(self) -> None: ...
    def kill(self) -> None: ...
    async def communicate(self, input: bytes | bytearray | memoryview | None = ...) -> tuple[bytes, bytes]: ...

if sys.version_info >= (3, 10):
    async def create_subprocess_shell(
        cmd: str | bytes,
        stdin: int | IO[Any] | None = ...,
        stdout: int | IO[Any] | None = ...,
        stderr: int | IO[Any] | None = ...,
        limit: int = ...,
        *,
        # These parameters are forced to these values by BaseEventLoop.subprocess_shell
        universal_newlines: Literal[False] = ...,
        shell: Literal[True] = ...,
        bufsize: Literal[0] = ...,
        encoding: None = ...,
        errors: None = ...,
        text: Literal[False, None] = ...,
        # These parameters are taken by subprocess.Popen, which this ultimately delegates to
        executable: StrOrBytesPath | None = ...,
        preexec_fn: Callable[[], Any] | None = ...,
        close_fds: bool = ...,
        cwd: StrOrBytesPath | None = ...,
        env: subprocess._ENV | None = ...,
        startupinfo: Any | None = ...,
        creationflags: int = ...,
        restore_signals: bool = ...,
        start_new_session: bool = ...,
        pass_fds: Any = ...,
    ) -> Process: ...
    async def create_subprocess_exec(
        program: _ExecArg,
        *args: _ExecArg,
        stdin: int | IO[Any] | None = ...,
        stdout: int | IO[Any] | None = ...,
        stderr: int | IO[Any] | None = ...,
        limit: int = ...,
        # These parameters are forced to these values by BaseEventLoop.subprocess_shell
        universal_newlines: Literal[False] = ...,
        shell: Literal[True] = ...,
        bufsize: Literal[0] = ...,
        encoding: None = ...,
        errors: None = ...,
        # These parameters are taken by subprocess.Popen, which this ultimately delegates to
        text: bool | None = ...,
        executable: StrOrBytesPath | None = ...,
        preexec_fn: Callable[[], Any] | None = ...,
        close_fds: bool = ...,
        cwd: StrOrBytesPath | None = ...,
        env: subprocess._ENV | None = ...,
        startupinfo: Any | None = ...,
        creationflags: int = ...,
        restore_signals: bool = ...,
        start_new_session: bool = ...,
        pass_fds: Any = ...,
    ) -> Process: ...

else:
    async def create_subprocess_shell(
        cmd: str | bytes,
        stdin: int | IO[Any] | None = ...,
        stdout: int | IO[Any] | None = ...,
        stderr: int | IO[Any] | None = ...,
        loop: events.AbstractEventLoop | None = ...,
        limit: int = ...,
        *,
        # These parameters are forced to these values by BaseEventLoop.subprocess_shell
        universal_newlines: Literal[False] = ...,
        shell: Literal[True] = ...,
        bufsize: Literal[0] = ...,
        encoding: None = ...,
        errors: None = ...,
        text: Literal[False, None] = ...,
        # These parameters are taken by subprocess.Popen, which this ultimately delegates to
        executable: StrOrBytesPath | None = ...,
        preexec_fn: Callable[[], Any] | None = ...,
        close_fds: bool = ...,
        cwd: StrOrBytesPath | None = ...,
        env: subprocess._ENV | None = ...,
        startupinfo: Any | None = ...,
        creationflags: int = ...,
        restore_signals: bool = ...,
        start_new_session: bool = ...,
        pass_fds: Any = ...,
    ) -> Process: ...
    async def create_subprocess_exec(
        program: _ExecArg,
        *args: _ExecArg,
        stdin: int | IO[Any] | None = ...,
        stdout: int | IO[Any] | None = ...,
        stderr: int | IO[Any] | None = ...,
        loop: events.AbstractEventLoop | None = ...,
        limit: int = ...,
        # These parameters are forced to these values by BaseEventLoop.subprocess_shell
        universal_newlines: Literal[False] = ...,
        shell: Literal[True] = ...,
        bufsize: Literal[0] = ...,
        encoding: None = ...,
        errors: None = ...,
        # These parameters are taken by subprocess.Popen, which this ultimately delegates to
        text: bool | None = ...,
        executable: StrOrBytesPath | None = ...,
        preexec_fn: Callable[[], Any] | None = ...,
        close_fds: bool = ...,
        cwd: StrOrBytesPath | None = ...,
        env: subprocess._ENV | None = ...,
        startupinfo: Any | None = ...,
        creationflags: int = ...,
        restore_signals: bool = ...,
        start_new_session: bool = ...,
        pass_fds: Any = ...,
    ) -> Process: ...