Source code for datalad_next.runners.git

from __future__ import annotations

from pathlib import Path
import subprocess

from datalad_next.exceptions import CapturedException

from .iter_subproc import (
    CommandError,
    iter_subproc,
)


def _call_git(
    args: list[str],
    *,
    capture_output: bool = False,
    cwd: Path | None = None,
    check: bool = False,
    text: bool | None = None,
    # TODO
    #patch_env: dict[str, str] | None = None,
) -> subprocess.CompletedProcess:
    """Wrapper around ``subprocess.run`` for calling Git command

    ``args`` is a list of argument for the Git command. This list must not
    contain the Git executable itself. It will be prepended (unconditionally)
    to the arguments before passing them on.

    All other argument are pass on to ``subprocess.run()`` verbatim.
    """
    # make configurable
    git_executable = 'git'
    cmd = [git_executable, *args]
    try:
        return subprocess.run(
            cmd,
            capture_output=capture_output,
            cwd=cwd,
            check=check,
            text=text,
        )
    except subprocess.CalledProcessError as e:
        # TODO we could support post-error forensics, but some client
        # might call this knowing that it could fail, and may not
        # appreciate the slow-down. Add option `expect_fail=False`?
        #
        # normalize exception to datalad-wide standard
        raise CommandError(
            cmd=cmd,
            code=e.returncode,
            stdout=e.stdout,
            stderr=e.stderr,
            cwd=cwd,
        ) from e


def call_git(
    args: list[str],
    *,
    cwd: Path | None = None,
) -> None:
    """Call git with no output capture, raises on non-zero exit.

    If ``cwd`` is not None, the function changes the working directory to
    ``cwd`` before executing the command.
    """
    _call_git(
        args,
        capture_output=False,
        cwd=cwd,
        check=True,
    )


def call_git_success(
    args: list[str],
    *,
    cwd: Path | None = None,
) -> bool:
    """Call Git for a single line of output.

    ``args`` is a list of arguments for the Git command. This list must not
    contain the Git executable itself. It will be prepended (unconditionally)
    to the arguments before passing them on.

    If ``cwd`` is not None, the function changes the working directory to
    ``cwd`` before executing the command.
    """
    try:
        _call_git(
            args,
            capture_output=False,
            cwd=cwd,
            check=True,
        )
    except CommandError as e:
        CapturedException(e)
        return False
    return True


def call_git_lines(
    args: list[str],
    *,
    cwd: Path | None = None,
) -> bool:
    """Call Git for any (small) number of lines of output.

    ``args`` is a list of arguments for the Git command. This list must not
    contain the Git executable itself. It will be prepended (unconditionally)
    to the arguments before passing them on.

    If ``cwd`` is not None, the function changes the working directory to
    ``cwd`` before executing the command.

    Raises
    ------
    CommandError if the call exits with a non-zero status.
    """
    res = _call_git(
        args,
        capture_output=True,
        cwd=cwd,
        check=True,
        text=True,
    )
    return res.stdout.splitlines()


def call_git_oneline(
    args: list[str],
    *,
    cwd: Path | None = None,
) -> str:
    """Call git for a single line of output.

    If ``cwd`` is not None, the function changes the working directory to
    ``cwd`` before executing the command.

    Raises
    ------
    CommandError if the call exits with a non-zero status.
    AssertionError if there is more than one line of output.
    """
    lines = call_git_lines(args, cwd=cwd)
    if len(lines) > 1:
        raise AssertionError(
            f"Expected Git {args} to return a single line, but got {lines}"
        )
    return lines[0]


[docs] def iter_git_subproc( args: list[str], **kwargs ): """``iter_subproc()`` wrapper for calling Git commands All argument semantics are identical to those of ``iter_subproc()``, except that ``args`` must not contain the Git binary, but need to be exclusively arguments to it. The respective `git` command/binary is automatically added internally. """ cmd = ['git'] cmd.extend(args) return iter_subproc(cmd, **kwargs)