Add pipesize argument to subprocess.{run, call, check_call, check_output} (#8124)

This commit is contained in:
Alex Waygood
2022-06-21 13:59:26 +01:00
committed by GitHub
parent cad42c7c7b
commit a01e0260b3

View File

@@ -101,7 +101,209 @@ class CompletedProcess(Generic[_T]):
if sys.version_info >= (3, 9):
def __class_getitem__(cls, item: Any) -> GenericAlias: ...
if sys.version_info >= (3, 9):
if sys.version_info >= (3, 10):
# 3.10 adds "pipesize" argument
@overload
def run(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stdout: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
universal_newlines: bool = ...,
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
*,
capture_output: bool = ...,
check: bool = ...,
encoding: str | None = ...,
errors: str | None = ...,
input: str | None = ...,
text: Literal[True],
timeout: float | None = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> CompletedProcess[str]: ...
@overload
def run(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stdout: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
universal_newlines: bool = ...,
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
*,
capture_output: bool = ...,
check: bool = ...,
encoding: str,
errors: str | None = ...,
input: str | None = ...,
text: bool | None = ...,
timeout: float | None = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> CompletedProcess[str]: ...
@overload
def run(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stdout: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
universal_newlines: bool = ...,
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
*,
capture_output: bool = ...,
check: bool = ...,
encoding: str | None = ...,
errors: str,
input: str | None = ...,
text: bool | None = ...,
timeout: float | None = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> CompletedProcess[str]: ...
@overload
def run(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stdout: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
*,
universal_newlines: Literal[True],
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
# where the *real* keyword only args start
capture_output: bool = ...,
check: bool = ...,
encoding: str | None = ...,
errors: str | None = ...,
input: str | None = ...,
text: bool | None = ...,
timeout: float | None = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> CompletedProcess[str]: ...
@overload
def run(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stdout: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
universal_newlines: Literal[False] = ...,
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
*,
capture_output: bool = ...,
check: bool = ...,
encoding: None = ...,
errors: None = ...,
input: bytes | None = ...,
text: Literal[None, False] = ...,
timeout: float | None = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> CompletedProcess[bytes]: ...
@overload
def run(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stdout: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
universal_newlines: bool = ...,
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
*,
capture_output: bool = ...,
check: bool = ...,
encoding: str | None = ...,
errors: str | None = ...,
input: _TXT | None = ...,
text: bool | None = ...,
timeout: float | None = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> CompletedProcess[Any]: ...
elif sys.version_info >= (3, 9):
# 3.9 adds arguments "user", "group", "extra_groups" and "umask"
@overload
def run(
@@ -604,7 +806,37 @@ else:
) -> CompletedProcess[Any]: ...
# Same args as Popen.__init__
if sys.version_info >= (3, 9):
if sys.version_info >= (3, 10):
# 3.10 adds "pipesize" argument
def call(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stdout: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
universal_newlines: bool = ...,
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
*,
timeout: float | None = ...,
text: bool | None = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> int: ...
elif sys.version_info >= (3, 9):
# 3.9 adds arguments "user", "group", "extra_groups" and "umask"
def call(
args: _CMD,
@@ -682,7 +914,37 @@ else:
) -> int: ...
# Same args as Popen.__init__
if sys.version_info >= (3, 9):
if sys.version_info >= (3, 10):
# 3.10 adds "pipesize" argument
def check_call(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath = ...,
stdin: _FILE = ...,
stdout: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
universal_newlines: bool = ...,
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
timeout: float | None = ...,
*,
text: bool | None = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> int: ...
elif sys.version_info >= (3, 9):
# 3.9 adds arguments "user", "group", "extra_groups" and "umask"
def check_call(
args: _CMD,
@@ -758,7 +1020,191 @@ else:
timeout: float | None = ...,
) -> int: ...
if sys.version_info >= (3, 9):
if sys.version_info >= (3, 10):
# 3.10 adds "pipesize" argument
@overload
def check_output(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
universal_newlines: bool = ...,
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
*,
timeout: float | None = ...,
input: _TXT | None = ...,
encoding: str | None = ...,
errors: str | None = ...,
text: Literal[True],
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> str: ...
@overload
def check_output(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
universal_newlines: bool = ...,
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
*,
timeout: float | None = ...,
input: _TXT | None = ...,
encoding: str,
errors: str | None = ...,
text: bool | None = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> str: ...
@overload
def check_output(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
universal_newlines: bool = ...,
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
*,
timeout: float | None = ...,
input: _TXT | None = ...,
encoding: str | None = ...,
errors: str,
text: bool | None = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> str: ...
@overload
def check_output(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
*,
universal_newlines: Literal[True],
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
# where the real keyword only ones start
timeout: float | None = ...,
input: _TXT | None = ...,
encoding: str | None = ...,
errors: str | None = ...,
text: bool | None = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> str: ...
@overload
def check_output(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
universal_newlines: Literal[False] = ...,
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
*,
timeout: float | None = ...,
input: _TXT | None = ...,
encoding: None = ...,
errors: None = ...,
text: Literal[None, False] = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> bytes: ...
@overload
def check_output(
args: _CMD,
bufsize: int = ...,
executable: StrOrBytesPath | None = ...,
stdin: _FILE = ...,
stderr: _FILE = ...,
preexec_fn: Callable[[], Any] | None = ...,
close_fds: bool = ...,
shell: bool = ...,
cwd: StrOrBytesPath | None = ...,
env: _ENV | None = ...,
universal_newlines: bool = ...,
startupinfo: Any = ...,
creationflags: int = ...,
restore_signals: bool = ...,
start_new_session: bool = ...,
pass_fds: Any = ...,
*,
timeout: float | None = ...,
input: _TXT | None = ...,
encoding: str | None = ...,
errors: str | None = ...,
text: bool | None = ...,
user: str | int | None = ...,
group: str | int | None = ...,
extra_groups: Iterable[str | int] | None = ...,
umask: int = ...,
pipesize: int = ...,
) -> Any: ... # morally: -> _TXT
elif sys.version_info >= (3, 9):
# 3.9 adds arguments "user", "group", "extra_groups" and "umask"
@overload
def check_output(