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

Bases: Exception

__init__(output, *args)

Parameters:

• output (str)

• args (object)

Return type:

None

libvcs._internal.subprocess._CMD: TypeAlias = str | bytes | os.PathLike[str] | os.PathLike[bytes] | collections.abc.Sequence[str | bytes | os.PathLike[str] | os.PathLike[bytes]]

Command

class libvcs._internal.subprocess.SubprocessCommand

Bases: SkipDefaultFieldsReprMixin

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

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: str | bytes | PathLike[str] | PathLike[bytes] | Sequence[str | bytes | PathLike[str] | PathLike[bytes]]

bufsize: int = -1

executable: str | bytes | PathLike[str] | PathLike[bytes] | None = None

stdin: None | int | IO[Any] = None

stdout: None | int | IO[Any] = None

stderr: None | int | IO[Any] = None

preexec_fn: Callable[[], Any] | None = None

close_fds: bool = True

shell: bool = False

cwd: str | bytes | PathLike[str] | PathLike[bytes] | None = None

env: Mapping[bytes, str | bytes | PathLike[str] | PathLike[bytes]] | Mapping[str, str | bytes | PathLike[str] | PathLike[bytes]] | None = None

creationflags: int = 0

startupinfo: Any | None = None

restore_signals: bool = True

start_new_session: bool = False

pass_fds: Any = ()

umask: int = -1

pipesize: int = -1

user: str | None = None

group: str | None = None

extra_groups: list[str] | None = None

universal_newlines: bool | None = None

text: Literal[True] | None = None

encoding: str | None = None

errors: str | None = None

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

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)

• stdin (None | int | IO[Any])

• stdout (None | int | IO[Any])

• stderr (None | int | IO[Any])

• preexec_fn (Callable[[], Any] | None)

• close_fds (bool)

• shell (bool)

• cwd (str | bytes | PathLike[str] | PathLike[bytes] | None)

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

• extra_groups (list[str] | None)

• universal_newlines (bool | None)

• text (Literal[True] | None)

• encoding (str | None)

• errors (str | None)

Return type:

None

Popen(args=None, universal_newlines=None, *, text=None, encoding=None, errors=None, **kwargs)

Run commands subprocess.Popen, optionally overrides via kwargs.

Parameters:

• **kwargs (Any) – Overrides existing attributes for subprocess.Popen

• args (str | bytes | PathLike[str] | PathLike[bytes] | Sequence[str | bytes | PathLike[str] | PathLike[bytes]] | None)

• universal_newlines (bool | None)

• text (bool | None)

• encoding (str | None)

• errors (str | None)

• **kwargs

Return type:

Popen[Any]

Examples

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

check_call(**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(universal_newlines=None, *, input=None, encoding=None, errors=None, text=None, **kwargs)

Run command subprocess.check_output(), optionally override via kwargs.

Parameters:

• input (t.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:

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)

Run command in subprocess.run(), optionally overrides via kwargs.

Parameters:

• input (t.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''