Command helpers - libvcs._internal.run

Misc. legacy helpers subprocess and finding VCS binaries.

libvcs._internal.run.run will be deprecated by libvcs._internal.subprocess.

Note

This is an internal API not covered by versioning policy.

libvcs._internal.run.console_to_str(s)

function

From pypa/pip project, pip.backwardwardcompat. License MIT.

Parameters:

s (bytes)

Return type:

str

class libvcs._internal.run.CmdLoggingAdapter

class

Bases: LoggerAdapter

Adapter for additional command-related data to logging.

Extends logging.LoggerAdapter’s functionality.

Mixes in additional context via logging.LoggerAdapter.process() for logging.Formatter when emitting log entries.

Parameters:

• bin_name (str) – name of the command or vcs tool being wrapped, e.g. ‘git’

• keyword (str) – directory basename, name of repo, hint, etc. e.g. ‘django’

__init__(bin_name, keyword, *args, **kwargs)

method

Initialize the adapter with a logger and an optional dict-like object which provides contextual information. This constructor signature allows easy stacking of LoggerAdapters, if so desired.

You can effectively pass keyword arguments as shown in the following example:

adapter = LoggerAdapter(someLogger, dict(p1=v1, p2=”v2”))

By default, LoggerAdapter objects will drop the “extra” argument passed on the individual log calls to use its own instead.

Initializing it with merge_extra=True will instead merge both maps when logging, the individual call extra taking precedence over the LoggerAdapter instance extra

Changed in version 3.13: The merge_extra argument was added.

Parameters:

• bin_name (str)

• keyword (str)

• args (Any)

• kwargs (Any)

Return type:

None

bin_name

attribute

bin_name

keyword

attribute

directory basename, name of repository, hint, etc.

process(msg, kwargs)

method

Add additional context information for loggers.

Parameters:

• msg (str)

• kwargs (MutableMapping[str, Any])

Return type:

tuple[Any, MutableMapping[str, Any]]

class libvcs._internal.run.ProgressCallbackProtocol

class

Bases: Protocol

Callback to report subprocess communication.

__init__(*args, **kwargs)

method

_abc_impl = <_abc._abc_data object>

attribute

_is_protocol = True

attribute

libvcs._internal.run._normalize_command_args(args)

function

Return subprocess arguments without splitting scalar strings or bytes.

Parameters:

args (_CMD)

Return type:

list[StrOrBytesPath]

libvcs._internal.run._stringify_command(args)

function

Return a human-readable command for CommandError.

Parameters:

args (_CMD)

Return type:

str | list[str]

libvcs._internal.run.run(args, bufsize=-1, executable=None, stdin=None, stdout=None, stderr=None, preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, startupinfo=None, creationflags=0, restore_signals=True, start_new_session=False, pass_fds=(), *, encoding=None, errors=None, user=None, group=None, extra_groups=None, umask=-1, log_in_real_time=False, check_returncode=True, callback=None)

function

Run a command.

Run ‘args’ in a shell and return the combined contents of stdout and stderr (Blocking). Throws an exception if the command exits non-zero.

Keyword arguments are passthrough to subprocess.Popen.

Parameters:

• args (list or str, or single str, if shell=True) – the command to run

• shell (bool) – boolean indicating whether we are using advanced shell features. Use only when absolutely necessary, since this allows a lot more freedom which could be exploited by malicious code. See the warning here: http://docs.python.org/library/subprocess.html#popen-constructor

• cwd (str) – dir command is run from. Defaults to path.

• log_in_real_time (bool) – boolean indicating whether to read stdout from the subprocess in real time instead of when the process finishes.

• check_returncode (bool) – Indicate whether a libvcs.exc.CommandError should be raised if return code is different from 0.

• callback (ProgressCallbackProtocol) –

  callback to return output as a command executes, accepts a function signature of (output, timestamp). Example usage:

  def progress_cb(output, timestamp):
      sys.stdout.write(output)
      sys.stdout.flush()
  run(['git', 'pull'], callback=progress_cb)
  

• changes (Upcoming)

• ----------------

• 3.10 (When minimum python >=)

• pipesize (int = -1 will be added after umask.)

• bufsize (int)

• executable (StrOrBytesPath | None)

• stdin (_FILE | None)

• stdout (_FILE | None)

• stderr (_FILE | None)

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

• close_fds (bool)

• env (_ENV | None)

• startupinfo (Any | None)

• creationflags (int)

• restore_signals (bool)

• start_new_session (bool)

• pass_fds (Any)

• encoding (str | None)

• errors (str | None)

• user (str | int | None)

• group (str | int | None)

• extra_groups (Iterable[str | int] | None)

• umask (int)

Return type:

str