`libvcs._internal.subprocess`#

Invokable 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: Wraps subprocess.Popen and subprocess.run() in a dataclass().

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'

class libvcs._internal.subprocess.SubprocessCommand#

Encapsulate a subprocess request. Inspect, mutate, control before invocation.

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:: int

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

Popen(self, args: Optional[_CMD] = ..., universal_newlines: bool = ..., *, text: Optional[bool] = ..., encoding: str, errors: Optional[str] = ...)#

Run commands subprocess.Popen, optionally overrides via kwargs.

Parameters:: **kwargs (dict, optional) – Overrides existing attributes for subprocess.Popen

Examples

>>> cmd = SubprocessCommand(args=['echo', 'hello'])
>>> proc = cmd.Popen(stdout=subprocess.PIPE)
>>> proc.communicate() 

check_call(self, **kwargs)#

Run command subprocess.check_call(), optionally overrides via kwargs.

Parameters:: **kwargs (dict, optional) – Overrides existing attributes for subprocess.check_call()
Return type:: int

Examples

>>> cmd = SubprocessCommand(args=['echo', 'hello'])
>>> cmd.check_call(stdout=subprocess.PIPE)
0

check_output(self, universal_newlines: bool = ..., *, input: Optional[Union[str, bytes]] = ..., encoding: Optional[str] = ..., errors: Optional[str] = ..., text: Literal[True], **kwargs)#

Run command subprocess.check_output(), 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.
**kwargs (dict, optional) – Overrides existing attributes for subprocess.check_output()

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(self, universal_newlines: bool = ..., *, capture_output: bool = ..., check: bool = ..., encoding: Optional[str] = ..., errors: Optional[str] = ..., input: Optional[str] = ..., text: Literal[True])#

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()

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''

libvcs._internal.subprocess#

`libvcs._internal.subprocess`#