SubprocessCommand - 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'
libvcs._internal.subprocess._CMD#

Command

alias of Union[str, bytes, PathLike[str], PathLike[bytes], Sequence[Union[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#

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

Type:

Union[None, int, IO[Any]]

stderr#

standard output for executed program

Type:

Union[None, int, IO[Any]]

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.

Type:

Optional[Literal[True]]

universal_newlines#

Alias of text, provided for backwards compatibility.

Type:

Optional[bool]

startupinfo#

Windows only

Type:

Optional[Any]

creationflags#

Windows only

Type:

int

preexec_fn#

(POSIX only) An object to be called in the child process just before the child is executed.

Type:

Optional[Callable[[], Any]]

restore_signals#

POSIX only

Type:

bool

start_new_session#

POSIX only

Type:

bool

group#

POSIX only

Type:

Optional[str]

extra_groups#

POSIX only

Type:

Optional[list[str]]

user#

POSIX only

Type:

Optional[str]

umask#

POSIX only

Type:

int

pass_fds#

POSIX only

Type:

Any

encoding#

Text mode encoding to use for file objects stdin, stdout and stderr.

Type:

Optional[str]

errors#

Text mode error handling to use for file objects stdin, stdout and stderr.

Type:

Optional[str]

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]]]]#
bufsize: int = -1#
executable: Optional[Union[str, bytes, PathLike[str], PathLike[bytes]]] = None#
stdin: Union[None, int, IO[Any]] = None#
stdout: Union[None, int, IO[Any]] = None#
stderr: Union[None, int, IO[Any]] = None#
preexec_fn: Optional[Callable[[], Any]] = None#
close_fds: bool = True#
shell: bool = False#
cwd: Optional[Union[str, bytes, PathLike[str], PathLike[bytes]]] = None#
env: Optional[Union[Mapping[bytes, Union[str, bytes, PathLike[str], PathLike[bytes]]], Mapping[str, Union[str, bytes, PathLike[str], PathLike[bytes]]]]] = None#
creationflags: int = 0#
startupinfo: Optional[Any] = None#
restore_signals: bool = True#
start_new_session: bool = False#
pass_fds: Any = ()#
umask: int = -1#
pipesize: int = -1#
user: Optional[str] = None#
group: Optional[str] = None#
extra_groups: Optional[list[str]] = None#
universal_newlines: Optional[bool] = None#
text: Optional[Literal[True]] = None#
encoding: Optional[str] = None#
errors: Optional[str] = None#
Popen(args: Optional[Union[str, bytes, PathLike[str], PathLike[bytes], Sequence[Union[str, bytes, PathLike[str], PathLike[bytes]]]]] = None, universal_newlines: bool = None, *, text: Optional[bool] = None, encoding: str, errors: Optional[str] = None) → Popen[str][source]#
Popen(args: Optional[Union[str, bytes, PathLike[str], PathLike[bytes], Sequence[Union[str, bytes, PathLike[str], PathLike[bytes]]]]] = None, universal_newlines: bool = None, *, text: Optional[bool] = None, encoding: Optional[str] = None, errors: str) → Popen[str]
Popen(args: Optional[Union[str, bytes, PathLike[str], PathLike[bytes], Sequence[Union[str, bytes, PathLike[str], PathLike[bytes]]]]] = None, *, universal_newlines: Literal[True], text: Optional[bool] = None, encoding: Optional[str] = None, errors: Optional[str] = None) → Popen[str]
Popen(args: Optional[Union[str, bytes, PathLike[str], PathLike[bytes], Sequence[Union[str, bytes, PathLike[str], PathLike[bytes]]]]] = None, universal_newlines: bool = None, *, text: Literal[True], encoding: Optional[str] = None, errors: Optional[str] = None) → Popen[str]
Popen(args: Optional[Union[str, bytes, PathLike[str], PathLike[bytes], Sequence[Union[str, bytes, PathLike[str], PathLike[bytes]]]]] = None, universal_newlines: Literal[False] = None, *, text: Literal[None, False] = None, encoding: None = None, errors: None = None) → Popen[bytes]

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()
Return type:

Popen[Any]

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:

int

Examples

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

int

Parameters:

kwargs (Any) –

check_output(universal_newlines: bool = None, *, input: Optional[Union[str, bytes]] = None, encoding: Optional[str] = None, errors: Optional[str] = None, text: Literal[True], **kwargs: Any) → str[source]#
check_output(universal_newlines: Optional[bool] = None, *, input: Optional[Union[str, bytes]] = None, encoding: str, errors: Optional[str] = None, text: Optional[bool] = None, **kwargs: Any) → str
check_output(universal_newlines: bool = None, *, input: Optional[Union[str, bytes]] = None, encoding: Optional[str] = None, errors: str, text: Optional[bool] = None, **kwargs: Any) → str
check_output(universal_newlines: Literal[True] = None, *, input: Optional[Union[str, bytes]] = None, encoding: Optional[str] = None, errors: Optional[str] = None, text: Optional[bool] = None, **kwargs: Any) → str
check_output(universal_newlines: Literal[False], *, input: Optional[Union[str, bytes]] = None, encoding: None = None, errors: None = None, text: Literal[None, False] = None, **kwargs: Any) → bytes

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 (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'
Return type:

Union[bytes, str]

run(universal_newlines: bool = None, *, capture_output: bool = False, check: bool = False, encoding: Optional[str] = None, errors: Optional[str] = None, input: Optional[str] = None, text: Literal[True]) → CompletedProcess[str][source]#
run(universal_newlines: bool = None, *, capture_output: bool = False, check: bool = False, encoding: str, errors: Optional[str] = None, input: Optional[str] = None, text: Optional[bool] = None) → CompletedProcess[str]
run(universal_newlines: bool = None, *, capture_output: bool = False, check: bool = False, encoding: Optional[str] = None, errors: str, input: Optional[str] = None, text: Optional[bool] = None) → CompletedProcess[str]
run(*, universal_newlines: Literal[True], capture_output: bool = False, check: bool = False, encoding: Optional[str] = None, errors: Optional[str] = None, input: Optional[str] = None, text: Optional[bool] = None) → CompletedProcess[str]
run(universal_newlines: Literal[False] = None, *, capture_output: bool = False, check: bool = False, encoding: None = None, errors: None = None, input: Optional[bytes] = None, text: Literal[None, False] = None) → CompletedProcess[bytes]

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''
Return type:

CompletedProcess[Any]