SubprocessCommand - libvcs._internal.subprocess
¶
Invocable subprocess
wrapper.
Defer running a subprocess, such as by handing to an executor.
Note
This is an internal API not covered by versioning policy.
Examples
SubprocessCommand
: Wrapssubprocess.Popen
andsubprocess.run()
in adataclass()
.Before:
>>> import subprocess >>> subprocess.run( ... ['echo', 'hi'], ... capture_output=True, universal_newlines=True ... ).stdout 'hi\n'
With this:
>>> cmd = SubprocessCommand(['echo', 'hi']) >>> cmd.args ['echo', 'hi'] >>> cmd.run(capture_output=True, universal_newlines=True).stdout 'hi\n'
Tweak params before invocation:
>>> cmd = SubprocessCommand(['echo', 'hi']) >>> cmd.args[1] = 'hello' >>> cmd.args ['echo', 'hello'] >>> cmd.run(capture_output=True, universal_newlines=True).stdout 'hello\n'
- exception libvcs._internal.subprocess.SubprocessCheckOutputError(output, *args)[source]¶
Bases:
Exception
- libvcs._internal.subprocess._CMD¶
Command
alias of
str
|bytes
|PathLike
[str
] |PathLike
[bytes
] |Sequence
[str
|bytes
|PathLike
[str
] |PathLike
[bytes
]]
- class libvcs._internal.subprocess.SubprocessCommand(args, bufsize=-1, executable=None, stdin=None, stdout=None, stderr=None, preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, creationflags=0, startupinfo=None, restore_signals=True, start_new_session=False, pass_fds=(), umask=-1, pipesize=-1, user=None, group=None, extra_groups=None, universal_newlines=None, text=None, encoding=None, errors=None)[source]¶
Bases:
SkipDefaultFieldsReprMixin
Wraps a
subprocess
request. Inspect, mutate, control before invocation.- Parameters:
args (str | bytes | PathLike[str] | PathLike[bytes] | Sequence[str | bytes | PathLike[str] | PathLike[bytes]])
bufsize (int)
executable (str | bytes | PathLike[str] | PathLike[bytes] | None)
close_fds (bool)
shell (bool)
env (Mapping[bytes, str | bytes | PathLike[str] | PathLike[bytes]] | Mapping[str, str | bytes | PathLike[str] | PathLike[bytes]] | None)
creationflags (int)
startupinfo (Any | None)
restore_signals (bool)
start_new_session (bool)
pass_fds (Any)
umask (int)
pipesize (int)
user (str | None)
group (str | None)
universal_newlines (bool | None)
text (Literal[True] | None)
encoding (str | None)
errors (str | None)
- args¶
A string, or a sequence of program arguments.
- Type:
_CMD
- bufsize¶
supplied as the buffering argument to the open() function when creating the stdin/stdout/stderr pipe file objects
- Type:
- executable¶
A replacement program to execute.
- Type:
Optional[StrOrBytesPath]
- stdin¶
standard output for executed program
- Type:
_FILE
- stdout¶
standard output for executed program
- stderr¶
standard output for executed program
- close_fds¶
- Type:
Controls closing or inheriting of file descriptors.
- shell¶
- Type:
If true, the command will be executed through the shell.
- cwd¶
- Type:
Sets the current directory before the child is executed.
- env¶
- Type:
Defines the environment variables for the new process.
- text¶
If
True
, decode stdin, stdout and stderr using the given encoding (if set) or the system default otherwise.
- universal_newlines¶
Alias of text, provided for backwards compatibility.
- startupinfo¶
Windows only
- creationflags¶
Windows only
- preexec_fn¶
(POSIX only) An object to be called in the child process just before the child is executed.
- restore_signals¶
POSIX only
- start_new_session¶
POSIX only
- group¶
POSIX only
- extra_groups¶
POSIX only
- user¶
POSIX only
- umask¶
POSIX only
- pass_fds¶
POSIX only
- encoding¶
Text mode encoding to use for file objects stdin, stdout and stderr.
- errors¶
Text mode error handling to use for file objects stdin, stdout and stderr.
Examples
>>> cmd = SubprocessCommand("ls") >>> cmd.args 'ls'
With
shell=True
:>>> cmd = SubprocessCommand("ls -l", shell=True) >>> cmd.shell True >>> cmd.args 'ls -l' >>> cmd.check_call() 0
-
args:
Union
[str
,bytes
,PathLike
[str
],PathLike
[bytes
],Sequence
[Union
[str
,bytes
,PathLike
[str
],PathLike
[bytes
]]]]¶
-
env:
Union
[Mapping
[bytes
,Union
[str
,bytes
,PathLike
[str
],PathLike
[bytes
]]],Mapping
[str
,Union
[str
,bytes
,PathLike
[str
],PathLike
[bytes
]]],None
] = None¶
- Popen(args=None, universal_newlines=None, *, text=None, encoding=None, errors=None, **kwargs)[source]¶
Run commands
subprocess.Popen
, optionally overrides via kwargs.- Parameters:
- Return type:
Examples
>>> cmd = SubprocessCommand(args=['echo', 'hello']) >>> proc = cmd.Popen(stdout=subprocess.PIPE) >>> proc.communicate()
- check_call(**kwargs)[source]¶
Run command
subprocess.check_call()
, optionally overrides via kwargs.- Parameters:
**kwargs (dict, optional) – Overrides existing attributes for
subprocess.check_call()
- Return type:
Examples
>>> cmd = SubprocessCommand(args=['echo', 'hello']) >>> cmd.check_call(stdout=subprocess.PIPE) 0
- check_output(universal_newlines=None, *, input=None, encoding=None, errors=None, text=None, **kwargs)[source]¶
Run command
subprocess.check_output()
, optionally override via kwargs.- Parameters:
input (Union[bytes, str], optional) –
pass string to subprocess’s stdin. Bytes by default, str in text mode.
Text mode is triggered by setting any of text, encoding, errors or universal_newlines.
**kwargs (Any) – Overrides existing attributes for
subprocess.check_output()
universal_newlines (bool | None)
encoding (str | None)
errors (str | None)
text (bool | None)
**kwargs
- Return type:
Examples
>>> cmd = SubprocessCommand(args=['echo', 'hello']) >>> proc = cmd.check_output(shell=True)
Examples from
subprocess
:>>> import subprocess >>> cmd = SubprocessCommand( ... ["/bin/sh", "-c", "ls -l non_existent_file ; exit 0"]) >>> cmd.check_output(stderr=subprocess.STDOUT) b"ls: ...non_existent_file...: No such file or directory\n"
>>> cmd = SubprocessCommand(["sed", "-e", "s/foo/bar/"]) >>> cmd.check_output(input=b"when in the course of fooman events\n") b'when in the course of barman events\n'
- run(universal_newlines=None, *, capture_output=False, check=False, encoding=None, errors=None, input=None, text=None, timeout=None, **kwargs)[source]¶
Run command in
subprocess.run()
, optionally overrides via kwargs.- Parameters:
input (Union[bytes, str], optional) –
pass string to subprocess’s stdin. Bytes by default, str in text mode.
Text mode is triggered by setting any of text, encoding, errors or universal_newlines.
check (bool) – If True and the exit code was non-zero, it raises a
subprocess.CalledProcessError
. The CalledProcessError object will have the return code in the returncode attribute, and output & stderr attributes if those streams were captured.timeout (int) – If given, and the process takes too long, a
subprocess.TimeoutExpired
**kwargs (dict, optional) – Overrides existing attributes for
subprocess.run()
universal_newlines (bool | None)
capture_output (bool)
encoding (str | None)
errors (str | None)
text (bool | None)
- Return type:
Examples
>>> import subprocess >>> cmd = SubprocessCommand( ... ["/bin/sh", "-c", "ls -l non_existent_file ; exit 0"]) >>> cmd.run() CompletedProcess(args=['/bin/sh', '-c', 'ls -l non_existent_file ; exit 0'], returncode=0)
>>> import subprocess >>> cmd = SubprocessCommand( ... ["/bin/sh", "-c", "ls -l non_existent_file ; exit 0"]) >>> cmd.run(check=True) CompletedProcess(args=['/bin/sh', '-c', 'ls -l non_existent_file ; exit 0'], returncode=0)
>>> cmd = SubprocessCommand(["sed", "-e", "s/foo/bar/"]) >>> completed = cmd.run(input=b"when in the course of fooman events\n") >>> completed CompletedProcess(args=['sed', '-e', 's/foo/bar/'], returncode=0) >>> completed.stderr
>>> cmd = SubprocessCommand(["sed", "-e", "s/foo/bar/"]) >>> completed = cmd.run(input=b"when in the course of fooman events\n", ... capture_output=True) >>> completed CompletedProcess(args=['sed', '-e', 's/foo/bar/'], returncode=0, stdout=b'when in the course of barman events\n', stderr=b'') >>> completed.stdout b'when in the course of barman events\n' >>> completed.stderr b''