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: 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'
    
exception libvcs._internal.subprocess.SubprocessCheckOutputError(output, *args)[source]

Bases: Exception

Parameters:
Return type:

None

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

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
args: Union[str, bytes, PathLike[str], PathLike[bytes], Sequence[Union[str, bytes, PathLike[str], PathLike[bytes]]]]
bufsize: int = -1
executable: Union[str, bytes, PathLike[str], PathLike[bytes], None] = 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: Union[str, bytes, PathLike[str], PathLike[bytes], None] = None
env: Union[Mapping[bytes, Union[str, bytes, PathLike[str], PathLike[bytes]]], Mapping[str, Union[str, bytes, PathLike[str], PathLike[bytes]]], None] = 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=None, universal_newlines=None, *, text=None, encoding=None, errors=None, **kwargs)[source]

Run commands subprocess.Popen, optionally overrides via kwargs.

Parameters:
Return type:

Popen[Any]

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:

int

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:

Union[bytes, str]

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:

CompletedProcess[Any]

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