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, timeout=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)
  

• timeout (float, optional) – Seconds to wait before terminating the subprocess. When the deadline is exceeded the process is sent SIGTERM (then SIGKILL after a short grace period) and libvcs.exc.CommandTimeoutError is raised with any output collected so far. None (default) disables the deadline and preserves the legacy behaviour of blocking until the process exits.

• 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

libvcs._internal.run._TIMEOUT_KILL_GRACE_SECONDS = 0.5

data

Grace period after terminate() before escalating to kill().

libvcs._internal.run._TIMEOUT_POLL_INTERVAL_SECONDS = 0.1

data

Upper bound on the selectors.select() wait inside the deadline loop.

libvcs._internal.run._wait_with_deadline(proc, *, deadline, timeout, callback, cmd)

function

Wait for proc to exit, enforcing a wall-clock deadline.

Drains both stdout and stderr concurrently so a child that fills either kernel pipe buffer (~64 KiB on Linux) cannot deadlock waiting for libvcs to read its output. Uses selectors to unblock when either stream is readable, when the child exits, or when the per-iteration poll interval expires – whichever comes first.

When the deadline is exceeded the subprocess is reaped and libvcs.exc.CommandTimeoutError is raised with the bytes captured before the timeout. Otherwise the captured stdout/stderr are returned to the caller alongside the exit code so they can be reused without re- reading the now-drained pipes.

Returns:

(returncode, stdout_bytes, stderr_bytes). A byte buffer is None when the corresponding pipe could not be put into non- blocking mode (Windows pipes, unusual fd types) – in that case the caller should fall back to reading the pipe directly.

Return type:

tuple[int, bytes | None, bytes | None]

Parameters:

• proc (Popen[bytes])

• deadline (float)

• timeout (float)

• callback (ProgressCallbackProtocol | None)

• cmd (str | list[str])

Notes

The progress callback is invoked for stderr chunks only, matching the legacy non-timeout codepath. stdout is drained into the returned buffer to prevent the child from blocking on a full pipe, but its chunks are not forwarded to the callback in real time. Callers that want streaming stdout should redirect it themselves (e.g. stdout= to a file) rather than relying on callback.

libvcs._internal.run._join_buffer(buffers, stream)

function

Concatenate captured chunks for stream, or None if not tracked.

Parameters:

• buffers (dict[IO[bytes], list[bytes]])

• stream (IO[bytes] | None)

Return type:

bytes | None

libvcs._internal.run._terminate_process(proc, cmd)

function

Terminate proc gracefully, falling back to kill on the grace.

Parameters:

• proc (Popen[bytes])

• cmd (str | list[str])

Return type:

None

libvcs._internal.run._format_cmd_for_log(cmd)

function

Render cmd as a flat string for the vcs_cmd log extra.

Parameters:

cmd (str | list[str])

Return type:

str

libvcs._internal.run._drain_stream(stream)

function

Best-effort read of any remaining bytes from a subprocess pipe.

Parameters:

stream (IO[bytes] | None)

Return type:

bytes